Producer

  • 提前配置 Topic 

    1. @Service
    2. public class RoccketMqTest {
    4.  final Logger log = LoggerFactory.getLogger(this.getClass());
    6.  private final DefaultMQProducer producer =
    7.                      new DefaultMQProducer("Tmc_Topic_Group");
    8.  @PostConstruct
    9.  public void init() throws MQClientException {
    10.      producer.setNamesrvAddr("ns1:9876;ns2:9876");
    11.      producer.start();
    12.  }
    14.  public DefaultMQProducer getProducer() {
    15.      return producer;
    16.  }
    18.  @PreDestroy
    19.  public void destroy() {
    20.      producer.shutdown();
    21.  }
    22. }

Consumer

  • 配置 消费组 

    1. @Service
    2. public class RocketMqConsumer {
    4.   protected DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
    6.   @PostConstruct
    7.   public void init() throws MQClientException {
    9.       consumer.setNamesrvAddr("ns1:9876;ns2:9876");
    10.       consumer.setConsumerGroup("Fullinfo_Consumer_Group");
    11.       consumer.subscribe("fullinfo", "*");
    12.       consumer.setConsumeMessageBatchMaxSize(128);
    13.       consumer.registerMessageListener(new MessageListenerConcurrently() {
    14.           @Override
    15.           public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
    16.               try {
    17.                   for (MessageExt msg : msgs) {
    18.                       final String topic = msg.getTopic();
    19.                       final byte[] body = msg.getBody();
    20.                       System.out.println(topic + "\t" +
    21.                               new String(body, StandardCharsets.UTF_8) +
    22.                               "\t" + msg.getTags());
    23.                   }
    24.               } catch (Exception e) {
    25.                   return ConsumeConcurrentlyStatus.RECONSUME_LATER;
    26.               }
    27.               return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    28.           }
    29.       });
    30.       consumer.start();
    31.   }
    33.   @PreDestroy
    34.   public void destroy() {
    35.       System.out.println("===shutdown===");
    36.       consumer.shutdown();
    37.   }
    38. }

    针对Consumer做切面，进行日志记录

    1. public abstract class AbstractTraceMessageListenerConcurrently implements MessageListenerConcurrently {
    3.   @Override
    4.   public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
    5.       LoggerContext.init();
    6.       try {
    7.           return doConsumeMessage(msgs, context);
    8.       } finally {
    9.           LoggerContext.remove();
    10.       }
    11.   }
    13.   /**
    14.    * 消费信息
    15.    *
    16.    * @param msgs    msgs
    17.    * @param context 上下文
    18.    * @return {@link ConsumeConcurrentlyStatus}
    19.    */
    20.   public abstract ConsumeConcurrentlyStatus doConsumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context);
    21. }