RocketMQ实战

首先准备一个SpringBoot工程，确保能够正常运行

POM

    <!-- 通用的一些类,方法,数据结构等 -->
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-common</artifactId>
        <version>4.3.2</version>
    </dependency>
    <!-- 访问客户端 -->
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-client</artifactId>
        <version>4.3.2</version>
    </dependency>
        <!-- JSON序列化工具 -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.49</version>
        </dependency>

application.yml

#server.port=8080
#server.tomcat.uri-encoding=utf-8
server:
  port: 8080
  tomcat:
    uri-encoding: utf-8

apache:
  rocketmq:
    ###MQ服务器地址
    namesrvAddr: 10.3.133.160:9876
    producer:
      ###生产者分组
      producerGroup: myProducer
    ###消费者名称
    consumer:
      PushConsumer: myPushConsumer1
      consumerGroup: xxx

消费者

@Configuration
public class ConsumerListener {

    @Value("${apache.rocketmq.consumer.consumerGroup}")
    private String consumerGroup;
    @Value("${apache.rocketmq.namesrvAddr}")
    private String namesrvAddr;

    @Bean
    public DefaultMQPushConsumer defaultMQPushConsumer() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(namesrvAddr);
        try {
            //广播模式消费
            //consumer.setMessageModel(MessageModel.BROADCASTING);
            consumer.subscribe("TopicTest1", "*");

            // 如果是第一次启动，从队列头部开始消费
            // 如果不是第一次启动，从上次消费的位置继续消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            consumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> {
                try {
                    for (MessageExt messageExt : list) {
                        String messageBody = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
                        System.out.println("[Consumer] msgID(" + messageExt.getMsgId() + ") msgBody : " + messageBody);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            consumer.start();
            System.out.println("[Consumer 已启动]");
        } catch (Exception e) {
            e.printStackTrace();
        }
        return consumer;
    }
}

生产者

@Service
public class ProducerServiceImpl {
    @Value("${apache.rocketmq.producer.producerGroup}")
    private String producerGroup;

    @Value("${apache.rocketmq.namesrvAddr}")
    private String namesrvAddr;

    private DefaultMQProducer producer;

    @PostConstruct
    public void initProducer() {
        producer = new DefaultMQProducer(producerGroup);
        producer.setNamesrvAddr(namesrvAddr);
        producer.setRetryTimesWhenSendFailed(3);
        try {
            producer.start();
            System.out.println("[Producer 已启动]");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public String send(String topic, String tags, String msg) {
        SendResult result = null;
        try {
            Message message = new Message(topic, tags, msg.getBytes(RemotingHelper.DEFAULT_CHARSET));
            result = producer.send(message);
            System.out.println("[Producer] msgID(" + result.getMsgId() + ") " + result.getSendStatus());
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "{\"MsgId\":\"" + result.getMsgId() + "\"}";
    }

    @PreDestroy
    public void shutDownProducer() {
        if (producer != null) {
            producer.shutdown();
        }
    }
}

测试接口

@Controller
@RequestMapping(value = "/mq")
public class MqController {

    @Value("${test-name}")
    private String name;

    @Resource
    private Student student;

    @Autowired
    private ProducerServiceImpl producer;



    @RequestMapping(value = "/push",method = RequestMethod.GET)
    @ResponseBody
    @ApiOperation(value = "测试接口", notes = "自定义请求头sessionId，sessionId的值是登陆接口返回的")
    public String test(){

        producer.send("TopicTest1", "push", "测试消息");

        return "Hello Springbootxxxx1111222333444!!!"+student.getName();
    }
}