java 消息队列RocketMQ学习

准备工作

windows下安装部署RocketMQ (参考链接)

1 下载：http://rocketmq.apache.org/release_notes/release-notes-4.7.0/
选择 'Binary'进行下载，解压。（根据需要自行选定版本）

2 配置

系统环境变量配置:
  变量名：ROCKETMQ_HOME
  变量值：MQ解压路径\MQ文件夹名

3 启动

• 一：MQ文件夹\bin目录下，打开cmd命令框

执行`start mqnamesrv.cmd`，启动NAMESERVER。
成功后会弹出提示框，此框勿关闭。

执行`start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true`，启动BROKER。
成功后会弹出提示框，此框勿关闭。

提示：假如弹出提示框提示‘错误: 找不到或无法加载主类 xxxxxx’。打开runbroker.cmd，然后将‘%CLASSPATH%’加上英文双引号。保存并重新执行start语句。

%CLASSPATH% -> "%CLASSPATH%"

• 二：启动rocketmq插件

下载地址：https://github.com/apache/rocketmq-externals.git

`\rocketmq-externals\rocketmq-console`文件夹下，打开cmd命令框
执行`mvn clean package -Dmaven.test.skip=true`，编译生成

编译成功之后，Cmd进入`\rocketmq-externals\rocketmq-console\target`文件夹
执行`java -jar rocketmq-console-ng-1.0.1.jar`，启动`rocketmq-console-ng-1.0.1.jar`

• 三: 测试
  浏览器中输入127.0.0.1:配置端口(rocketmq-console的端口)，成功后即可查看。

SpringBoot 整合 RocketMQ（参考链接）

1，添加pom.xml依赖

 <!--注意： 这里的版本,要和部署在服务器上的版本号一致-->
 <dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.7.0</version>
 </dependency>

2，JmsConfig（配置类）
连接RocketMQ服务器配置类，这里为了方便直接写成常量。

public class JmsConfig {
    /**
     * Name Server 地址，因为是集群部署 所以有多个用 分号 隔开
     */
    public static final String NAME_SERVER = "localhost:9876";
    /**
     * 主题名称 主题一般是服务器设置好 而不能在代码里去新建topic
     * （ 如果没有创建好，生产者往该主题发送消息 会报找不到topic错误）
     */
    public static final String TOPIC = "topic_poke";
}

3、Producer (生产者)

@Component
public class Producer {
    private String producerGroup = "test_producer";
    private DefaultMQProducer producer;
    public Producer() {
        //示例生产者
        producer = new DefaultMQProducer(producerGroup);
        //不开启vip通道 开通口端口会减2
        producer.setVipChannelEnabled(false);
        //绑定name server
        producer.setNamesrvAddr(JmsConfig.NAME_SERVER);
        start();
    }
    /**
     * 对象在使用之前必须要调用一次，只能初始化一次
     */
    public void start() {
        try {
            this.producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
    public DefaultMQProducer getProducer() {
        return producer;
    }
    /**
     * 一般在应用上下文，使用上下文监听器，进行关闭
     */
    public void shutdown() {
        this.producer.shutdown();
    }
}

4、Consumer (消费者)

@Component
public class Consumer {
    /**
     * 消费者实体对象
     */
    private DefaultMQPushConsumer consumer;
    /**
     * 消费者组
     */
    public static final String CONSUMER_GROUP = "test_consumer";
    /**
     * 通过构造函数 实例化对象
     */
    public Consumer() 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) -> {
            try {
                for (Message msg : msgs) {
                    //消费者获取消息 这里只输出 不做后面逻辑处理
                    String body = new String(msg.getBody(), "utf-8");
                    System.out.printf("Consumer-获取消息-主题topic为=%s, 消费消息为=%s", msg.getTopic(), body);
                }
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        System.out.println("消费者 启动成功=======");
    }
}

5,Controller测试

@RestController
public class Controller {
    @Autowired
    private Producer producer;

    private List<String> pokeList;
    /**
     * 初始化消息
     */
    public Controller() {
        pokeList = new ArrayList<>();
        pokeList.add("小火龍");
        pokeList.add("皮卡丘");
        pokeList.add("卡比兽");
        pokeList.add("快龙");
        pokeList.add("美纳斯");
    }
    @GetMapping("/text/rocketmq")
    public Object callback() throws Exception {
        int i = 0;
        for (String s : pokeList) {
            //创建生产信息
            Message message = new Message(JmsConfig.TOPIC, "test_tag", (i++ + "poke:" + s).getBytes());
            //发送
            SendResult sendResult = producer.getProducer().send(message);
            System.out.printf("输出生产者信息=%s \n", sendResult);
        }
        return "成功";
    }
}

• 完整demo
• demo-rocketmq