springboot整合activeMQ

一、添加依赖

<!--        rocketMQ依赖-->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.4.0</version>
</dependency>
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-common</artifactId>
    <version>4.4.0</version>
</dependency>

二、配置文件增加

自定义配置文件信息

apache:
  rocketmq:
    producer:
      producerGroup: Producer
    consumer:
      PushConsumer: orderConsumer
    namesrvAddr: 192.168.56.66:9876

三、配置生产者

配置producer

@Component
public class RocketProducer {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    /**
     * 初始化
     **/
    @PostConstruct
    public void init() {
        //生产者的组名
        producer = new DefaultMQProducer(producerGroup);
        //指定NameServer地址，多个地址以 ; 隔开
        //如 producer.setNamesrvAddr("192.168.100.141:9876;192.168.100.142:9876;192.168.100.149:9876");
        producer.setNamesrvAddr(namesrvAddr);
        producer.setVipChannelEnabled(false);

        try {
            //Producer对象在使用之前必须要调用start初始化，只能初始化一次
            producer.start();
        } catch (Exception e) {
            logger.error("初始化生产者失败。", e);
        }

        // producer.shutdown();  一般在应用上下文，关闭的时候进行关闭，用上下文监听器
    }

    /**
     * 生产者的组名
     */
    @Value("${apache.rocketmq.producer.producerGroup}")
    private String producerGroup;

    /**
     * NameServer 地址
     */
    @Value("${apache.rocketmq.namesrvAddr}")
    private String namesrvAddr;

    private DefaultMQProducer producer;

    public DefaultMQProducer getProducer() {
        return this.producer;
    }
}

四、配置消费者

@Component
public class RocketConsumer {
    @PostConstruct
    public void init() {
        //消费者的组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        //指定NameServer地址，多个地址以 ; 隔开
        consumer.setNamesrvAddr(namesrvAddr);

        try {
            //设置consumer所订阅的Topic和Tag，*代表全部的Tag
            consumer.subscribe("testTopic", "*");

            //CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，跳过历史消息
            //CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

            //MessageListenerOrderly 这个是有序的
            //MessageListenerConcurrently 这个是无序的,并行的方式处理，效率高很多
            consumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> {
                try {
                    for (MessageExt messageExt : list) {
                        System.out.println("messageExt: " + messageExt); //输出消息内容

                        String messageBody = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);

                        System.out.println("消费响应：msgId : " + messageExt.getMsgId() + ",  msgBody : " + messageBody); //输出消息内容
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER; //稍后再试
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消费成功
            });

            consumer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 消费者的组名
     */
    @Value("${apache.rocketmq.consumer.PushConsumer}")
    private String consumerGroup;

    /**
     * NameServer 地址
     */
    @Value("${apache.rocketmq.namesrvAddr}")
    private String namesrvAddr;
}

五、队列的使用

@GetMapping("rocketProducer")
public Map<String, Object> rocketProducer(String message, String tag) throws UnsupportedEncodingException, InterruptedException, RemotingException, MQClientException, MQBrokerException {
    Map<String, Object> jsonMap = new HashMap<>(16);
    Message msg = new Message("testTopic", tag, message.getBytes(RemotingHelper.DEFAULT_CHARSET));
    SendResult result = rocketProducer.getProducer().send(msg);
    System.out.println("返回响应id：" + result.getMsgId() + ",发送状态：" + result.getSendStatus());
    CommonUtil.writeSuccessMap(jsonMap, "消息发送成功");
    return jsonMap;
}