check multiple topics in one batch (#744)

errors/errors.go

@@ -47,4 +47,5 @@ var (
 	ErrMessageEmpty      = errors.New("message is nil")
 	ErrNotRunning        = errors.New("producer not started")
 	ErrPullConsumer      = errors.New("pull consumer has not supported")
+	ErrMultipleTopics    = errors.New("the topic of the messages in one batch should be the same")
 )

producer/producer.go

@@ -103,6 +103,14 @@ func (p *defaultProducer) checkMsg(msgs ...*primitive.Message) error {
 	if len(msgs[0].Topic) == 0 {
 		return errors2.ErrTopicEmpty
 	}
+
+	topic := msgs[0].Topic
+	for _, msg := range msgs {
+		if msg.Topic != topic {
+			return errors2.ErrMultipleTopics
+		}
+	}
+
 	return nil
 }