安科网

SpringBoot2.X 整合 RocketMQ4.X

MojitoBlogs

MojitoBlogs

2020-02-17

关注 关注

开发生产者代码

第一步：创建很普通的 SpringBoot 项目

第二步：加入相关依赖

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

第三步：写代码

SpringBoot2.X 整合 RocketMQ4.X

PayProducer 类如下所示：

package net.xdclass.xdclassmq.jms;

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

@Component
public class PayProducer {

    private String producerGroup = "pay_group";

    private String nameServerAddr = "192.168.0.104:9876";

    private DefaultMQProducer producer;

    public PayProducer() {
        producer = new DefaultMQProducer(producerGroup);

        //指定NameServer地址，多个地址以 ; 隔开
        //如 producer.setNamesrvAddr("192.168.100.141:9876;192.168.100.142:9876;192.168.100.149:9876");

        producer.setNamesrvAddr(nameServerAddr);
        start();
    }

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

    /**
     * 对象在使用之前必须要调用一次，只能初始化一次
     */
    public void start() {
        try {
            this.producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
    
    /**
     * 一般在应用上下文，使用上下文监听器，进行关闭
     */
    public void shutdown() {
        this.producer.shutdown();
    }
}

PayController 类如下所示：

package net.xdclass.xdclassmq.controller;

import net.xdclass.xdclassmq.jms.PayProducer;
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.HashMap;

@RestController
public class PayController {

    @Autowired
    private PayProducer payProducer;

    private static  final String topic = "pay_test_topic";

    @RequestMapping("/api/v1/pay_cb")
    public Object callback(String text) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Message message = new Message(topic,"taga", ("hello rocketmq = "+text).getBytes() );
        SendResult sendResult = payProducer.getProducer().send(message);
        System.out.println(sendResult);
        return new HashMap<>();
    }
}

第四步：测试

通过可视化管理后台查看消息

SpringBoot2.X 整合 RocketMQ4.X

SpringBoot2.X 整合 RocketMQ4.X

Message对象

  • topic：主题名称
  • tag：标签，用于过滤
  • key：消息唯一标示，可以是业务字段组合
  • body：消息体，字节数组

注意：发送消息到 Broker 前，需要判断是否有此 Topic。启动 Broker 的时候，本地环境建议开启自动创建 Topic，生产环境建议关闭自动化创建 Topic。建议先手工创建 Topic，如果靠程序自动创建，然后再投递消息，会出现延迟情况。自动创建topic： autoCreateTopicEnable=true 无效原因：客户端版本要和服务端版本保持一致。

概念模型: 一个 Topic 下面对应多个 Queue，可以在创建 Topic 时指定，如订单类 Topic。

常见错误一

org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException:
sendDefaultImpl call timeout
原因：阿里云存在多网卡，rocketmq都会根据当前网卡选择一个IP使用，当你的机器有多块网卡时，很有可能会有问题。比如，我遇到的问题是我机器上有两个IP，一个公网IP，一个私网IP, 因此需要配置broker.conf 指定当前的公网ip, 然后重新启动broker 
新增配置：conf/broker.conf  (属性名称brokerIP1=broker所在的公网ip地址 )
新增这个配置：brokerIP1=120.76.62.13  

启动命令：nohup sh bin/mqbroker -n localhost:9876  -c ./conf/broker.conf &

常见错误二

MQClientException: No route info of this topic, TopicTest1
原因：Broker 禁止自动创建 Topic，且用户没有通过手工方式创建 此Topic, 或者broker和Nameserver网络不通
解决：
通过 sh bin/mqbroker -m  查看配置
autoCreateTopicEnable=true 则自动创建topic

Centos7关闭防火墙  systemctl stop firewalld

常见错误三

控制台查看不了数据，提示连接 10909错误

原因：Rocket默认开启了VIP通道，VIP通道端口为10911-2=10909

解决：阿里云安全组需要增加一个端口 10909

其他错误:

https://blog.csdn.net/qq_14853889/article/details/81053145
https://blog.csdn.net/wangmx1993328/article/details/81588217#%E5%BC%82%E5%B8%B8%E8%AF%B4%E6%98%8E
https://www.jianshu.com/p/bfd6d849f156
https://blog.csdn.net/wangmx1993328/article/details/81588217

开发消费者代码

接着上面的工程，直接上代码，PayConsumer 类如下所示：

package net.xdclass.xdclassmq.jms;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
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.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.util.List;

@Component
public class PayConsumer {
    private DefaultMQPushConsumer consumer;

    private String CONSUMER_GROUP = "pay_consumer_group";
    private String NAME_SERVER = "192.168.0.104:9876";
    private String TOPIC = "pay_test_topic";

    public PayConsumer() throws MQClientException {

        consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
        consumer.setNamesrvAddr(this.NAME_SERVER);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        consumer.subscribe(this.TOPIC, "*");

//        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
//            try {
//                Message msg = msgs.get(0);
//                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), new String(msgs.get(0).getBody()));
//                String topic = msg.getTopic();
//                String body = new String(msg.getBody(), "utf-8");
//                String tags = msg.getTags();
//                String keys = msg.getKeys();
//                System.out.println("topic=" + topic + ", tags=" + tags + ", keys=" + keys + ", msg=" + body);
//                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//            } catch (UnsupportedEncodingException e) {
//                e.printStackTrace();
//                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
//            }
//        });

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                try {
                    Message msg = msgs.get(0);
                    System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), new String(msgs.get(0).getBody()));

                    String topic = msg.getTopic();
                    String body = new String(msg.getBody(), "utf-8");
                    String tags = msg.getTags();
                    String keys = msg.getKeys();
                    System.out.println("topic=" + topic + ", tags=" + tags + ", keys=" + keys + ", msg=" + body);

                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (UnsupportedEncodingException e) {

                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
        });

        consumer.start();
        System.out.println("consumer start ...");
    }
}

注释掉的部分采用 Lambda 表达式写法，效果是一样的。

常见问题

1、Caused by: org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to <172.17.42.1:10911> failed 

2、com.alibaba.rocketmq.client.exception.MQClientException: Send [1] times, still failed, cost [1647]ms, Topic: TopicTest1, BrokersSent: [broker-a, null, null]

3、org.apache.rocketmq.client.exception.MQClientException: Send [3] times, still failed, cost [497]ms, Topic: TopicTest, BrokersSent: [Book-Air.local, 	MacBook-Air.local, MacBook-Air.local]
解决：多网卡问题处理
	1、设置producer:  producer.setVipChannelEnabled(false);
	2、编辑ROCKETMQ 配置文件：broker.conf（下列ip为自己的ip）
		namesrvAddr = 192.168.0.101:9876
		brokerIP1 = 192.168.0.101

4、DESC: service not available now, maybe disk full, CL:
	解决：修改启动脚本runbroker.sh，在里面增加一句话即可：		
	JAVA_OPT="${JAVA_OPT} -Drocketmq.broker.diskSpaceWarningLevelRatio=0.98"
	（磁盘保护的百分比设置成98%，只有磁盘空间使用率达到98%时才拒绝接收producer消息）
	
常见问题处理
	https://blog.csdn.net/sqzhao/article/details/54834761
	https://blog.csdn.net/mayifan0/article/details/67633729
	https://blog.csdn.net/a906423355/article/details/78192828

rocketmq

MojitoBlogs

MojitoBlogs

0 关注 0 粉丝 0 动态

关注 关注

相关推荐

深入理解RocketMQ延迟消息

延迟消息是实际开发中一个非常有用的功能，本文第一部分从整体上介绍秒级精度延迟消息的实现思路，在第二部分结合RocketMQ的延迟消息实现，进行细致的讲解，点出关键部分的源码。第三步介绍延迟消息与消息重试的关系。Broker判断这是一个延迟消息后，将其通过临

IT农场 2020-11-13

消息疯狂堆积！RocketMQ 出 Bug 了?

用过 MQ 的同学，可能会遇到过消息堆积的问题。而我最近也踩上了这个坑，但是发现结果竟然是这么一个意料之外的原因而导致的。那一晚月和风高，我正准备踏上回家的路，突然收到告警短信轰炸！于是乎赶回公司赶紧打开电脑，登上 RocketMQ 后台查看

LCFlxfldy 2020-08-17

RabbitMQ都写了，RocketMQ怎么能落下？

整体架构最近看到了我在Github上写的rabbitmq-examples陆续被人star了，就想着写个rocketmq-examples。对rabbitmq感兴趣的小伙伴可以看我之前的文章。下面把RocketMQ的各个特性简单介绍一下，这样在用的时候心里

ljcsdn 2020-07-27

(五)消息队列

此时，为了提高用户体验和吞吐量，其实可以异步地调用系统B、C、D的接口。所以，我们可以通过引入消息中间件，再系统A执行完后将数据写入到消息队列中直接返回，而BCD系统则负责监听执行。通过以上分析可知，消息中间件最佳选型为RocketMQ。

LCFlxfldy 2020-07-05

RocketMQ(2)---核心概念、特性、使用等

对于RocketMQ而言，感觉官方提供的东西还是可以的：https://github.com/apache/rocketmq/tree/master/docs/cn

lypgcs 2020-06-27

RocketMq producer 发送一条消息所经过的流程

　　RocketMq producer 在发送一条消息时候，从 producer --nameSrv -- Broker 中间经过了什么样子的数据交互。从NameSrv 中获取topic 配置的相关信息，比如 broker 地址，队列数 之类的。

陈晨软件五千言 2020-06-17

RocketMQ系列（三）消息的生产与消费

前面的章节，我们已经把RocketMQ的环境搭建起来了，是一个两主两从的异步集群。接下来，我们就看看怎么去使用RocketMQ，在使用之前，先要在NameServer中创建Topic，我们知道RocketMQ是基于Topic的消息队列，在生产者发送消息的时

qingyuerji 2020-06-14

RocketMQ系列（四）顺序消费

折腾了好长时间才写这篇文章，顺序消费，看上去挺好理解的，就是消费的时候按照队列中的顺序一个一个消费；而并发消费，则是消费者同时从队列中取消息，同时消费，没有先后顺序。RocketMQ也有这两种方式的实现，但是在实践的过程中，就是不能顺序消费，好不容易能够实

MojitoBlogs 2020-06-14

RocketMQ系列（五）广播与延迟消息

今天要给大家介绍RocketMQ中的两个功能，一个是“广播”，这个功能是比较基础的，几乎所有的mq产品都是支持这个功能的；另外一个是“延迟消费”，这个应该算是RocketMQ的特色功能之一了吧。广播是把消息发送给订阅了这个主题的所有消费者。首先，消费者端的

lypgcs 2020-06-14

RocketMQ系列（一）基本概念

RocketMQ是阿里出品的一款开源的消息中间件，让其声名大噪的就是它的事务消息的功能。在企业中，消息中间件选择使用RocketMQ的还是挺多的，这一系列的文章都是针对RocketMQ的，咱们先从RocketMQ的一些基本概念和环境的搭建开始聊起。Rock

陈晨软件五千言 2020-06-14

RocketMq 在Netty 下是如何进行消息封装传输

　　要了解rocketMq 需要知道 数据在 rocketMq 中 是如何进行传输，在底层的结构到底是一个什么亚子，这个需要我们对Netty 对字符编解码有一些了解。　　我们从生产者发送消息，broker 接收消息 为例，来开展底层对消息结构。　　1. 如

meilongwhpu 2020-06-13

RocketMQ

然后执行‘start mqnamesrv.cmd’，启动NAMESERVER。成功后会弹出提示框，此框勿关闭。#set it false if you don‘t want use dashboard.default true. #set the mess

陈晨软件五千言 2020-06-11

RocketMQ集群监控平台搭建

RocketMQ有一个对其扩展的开源项目incubator-rocketmq-externals，这个项目中有一个子模块叫rocketmq-console，这个便是管理控制台项目了，先将incubator-rocketmq-externals拉到本地，因为

qingyuerji 2020-06-09

RocketMQ系列（三）消息的生产与消费

前面的章节，我们已经把RocketMQ的环境搭建起来了，是一个两主两从的异步集群。接下来，我们就看看怎么去使用RocketMQ，在使用之前，先要在NameServer中创建Topic，我们知道RocketMQ是基于Topic的消息队列，在生产者发送消息的时

MojitoBlogs 2020-06-09

RocketMQ事务消息

RocketMQ事务消息是指应用本地事务和发送消息操作可以被定义到全局事务中，要么同时成功，要么同时失败。RocketMQ的事务消息提供类似 X/Open XA 的分布事务功能，通过事务消息能达到分布式事务的最终一致。Producer 已经把消息成功发送到

meilongwhpu 2020-06-08

聊一聊高并发高可用那些事 - Kafka篇

偶尔推出一个优惠活动时，请求量极速上升。看了几个招聘网站，提到较多的消息队列有：RabbitMQ、RocketMQ、Kafka 以及 Redis 的消息队列和发布订阅模式。在高可用环境中，Kafka 需要部署多台，避免 Kafka 宕机后，服务无法访问。K

meilongwhpu 2020-06-08

RocketMQ-Namesrv源码解析

Name Server是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。Broker部署相对复杂，Broker分为Master和Slave，一个Master可以对应多个Slave，但是一个Slave只能对应一个Master，Master和Slave

lypgcs 2020-06-07

深度剖析 Kafka/RocketMQ 顺序消息的一些坑

我觉得这个问题问得很频繁，而且非常经典，在这里我就以 Kafka 为例子，说说我对 Kafka 顺序消息的一些理解吧，如有理解不对的地方麻烦留言指点一下。针对以上第二点，下面盘点一下 Kafka 集群中有哪些意外情况会打乱消息的顺序。

MojitoBlogs 2020-06-04

RocketMQ源码 — 九、 RocketMQ延时消息

上一节消息重试里面提到了重试的消息可以被延时消费，其实除此之外，用户发送的消息也可以指定延时时间，然后在指定延时时间之后投递消息，然后被consumer消费。broker收到消息在准备将消息写入存储的时候，判断是延时消息则更改Message的topic为延

meilongwhpu 2020-05-30

rocketmq初识

通常一个消息队列需要掌握的知识点有Topic（主体）、Producer（生产者）、Consumer（消费者）、Queue（队列）、Delivery Semantics. 蛋疼的是不同的消息队列关于这些名词叫法不一样，含义也不是很精确。所以阿里起了一个项目O

LCFlxfldy 2020-05-29
MojitoBlogs

MojitoBlogs

W3CSchool教程
HTML 教程
CSS 教程
Bootstrap 教程
Javascript 教程
jQuery 教程
后端教程
C 教程
Java 教程
PHP 教程
Python 教程
Go 教程
移动开发
Android 教程
Swift 教程
Kotlin 教程
jQuery Mobile 教程
ionic 教程
关于我们
新闻动态
联系方式
招聘英才
安科实验室
帮助与反馈

安科网(Ancii)，中国第一极客网

Copyright © 2013 - 2019 Ancii.com

京ICP备18063983号 京公网安备11010802014868号