springboot 使用rocketmq 4.8.0

pom.xml

        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.8.0</version>
        </dependency>


        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>

MqController.java

package com.example.test11;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.List;

/**
 * @author  
 * @description
 * @date 2021/3/13 17:27
 */
@RestController
@Slf4j
public class MqController {

    @Autowired
    private RocketMQProducer producer;

    private List<String> mesList;

    public MqController() {
        mesList = new ArrayList<>();
        mesList.add("小小");
        mesList.add("爸爸");
        mesList.add("妈妈");
        mesList.add("爷爷");
        mesList.add("奶奶");
    }

    @RequestMapping("/mq/test")
    public Object test() throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        for (String s : mesList) {
            Message message = new Message(JmsConfig.TOPIC, "testtag", ("小小一家人的称谓:" + s).getBytes());
            SendResult sendResult = producer.getProducer().send(message);
            log.info("输出生产者信息={}", sendResult);
        }
        return "成功";
    }
}

RocketMQConsumer.java

package com.example.test11;

//import cn.ifeng1.learning.rocketmq.demo.config.JmsConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;

/**
 * @author  
 * @description
 * @date 2021/3/13 14:30
 */
@Component
@Slf4j
public class RocketMQConsumer {

    /**
     * 消费者实体对象
     */
    private DefaultMQPushConsumer consumer;

    /**
     * 消费者组
     */
    public static final String CONSUMER_GROUP = "test_consumer";

    public RocketMQConsumer() throws MQClientException {
        consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
        consumer.setNamesrvAddr(JmsConfig.NAME_SERVER);
        // 消费模式:一个新的订阅组第一次启动从队列的最后位置开始消费 后续再启动接着上次消费的进度开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 订阅主题和标签
        consumer.subscribe(JmsConfig.TOPIC, "*");
        // 注册消费的监听，并在此监听中消费消息，并返回消费的状态信息
        consumer.registerMessageListener((MessageListenerConcurrently)(msgs, context) -> {
            //msgs中只收集同一个topic， 同一个tag，并且key相同的msg
            //会把不同的消息分别放置到不同的队列中
            try {
                for (Message msg : msgs) {
                    String body = new String(msg.getBody(), "utf-8");
                    log.info("消费者获取消息-主题topic为={}， 消费消息为={}", msg.getTopic(), body);
                }
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        System.out.println("消费者，启动成功======");
    }
}

RocketMQProducer.java

package com.example.test11;

import lombok.Getter;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.springframework.stereotype.Component;

/**
 * @author  
 * @description
 * @date 2021/3/13 14:03
 */
@Component
@Getter
public class RocketMQProducer {

    private String producerGroup = "test_producer";

    private DefaultMQProducer producer;

    public RocketMQProducer() {
        producer = new DefaultMQProducer(producerGroup);
        //不开启vip通道 开通口端口会减2
        producer.setVipChannelEnabled(false);
        producer.setNamesrvAddr(JmsConfig.NAME_SERVER);
        start();
    }

    /**
     * 对象在使用之前必须要调用一次，只能初始化一次
     */
    private void start() {
        try {
            this.producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    /**
     * 一般在应用上下文，使用上下文监听器，进行关闭
     */
    public void shutdown() {
        this.producer.shutdown();
    }

}