实战,用案例来说话
前面已经说了JMS和RocketMQ一些概念和安装,下面使用SpringBoot来亲身操作一下.
生产者的操作
- SpringBoot项目创建完成,引入依赖是第一步:
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.3.0</version>
</dependency>
- 创建生产者是第二步,生产者必须依赖于生产组,而且需要指定nameServer
@Component
public class PayProducer {
/**
* 生产组,生产者必须在生产组内
*/
private String producerGroup = "pay_group";
/**
* 端口
*/
private String nameServer = "39.106.214.179:9876";
private DefaultMQProducer producer;
public PayProducer() {
producer = new DefaultMQProducer(producerGroup);
// 指定nameServer地址,多个地址之间以 ; 隔开
producer.setNamesrvAddr(nameServer);
start();
}
public DefaultMQProducer getProducer() {
return producer;
}
/**
* 对象在使用之前必须调用一次,并且只能初始化一次
*/
public void start() {
try {
this.producer.start();
} catch (MQClientException e) {
e.printStackTrace();
}
}
/**
* 一般在应用上下文,使用上下文监听器,进行关闭
*/
public void shutdown() {
producer.shutdown();
}
}
- 创建Controller进行测试发送消息,必须要指定topic,消息依赖于主题
@RestController
public class PayController {
@Autowired
private PayProducer payProducer;
/**
* topic,消息依赖于topic
*/
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 send = payProducer.getProducer().send(message);
System.out.println(send);
return new HashMap<>();
}
}
-
采坑记录
- 上面完成就可以启动项目了,访问之后报错了:
MQClientException: No route info of this topic, TopicTest1 这个的原因就是Broker禁止自动创建Topic且用户没有通过手动方式创建Topic, 或者是broker与Nameserver网络不通 解决: 使用手动创建Topic,在RocketMQ控制台的主题中创建就好,最主要的是指定topic name,如下图 出现创建不了的情况往下看 如果还出现这个问题,请关闭防火墙
-
这次说下上面可能创建不了的问题,前面说了安装开放安全组,这次就是因为rocketMQ虚拟的端口问题,需要开放10909,也就是说ECS最终开放的端口号: 8080,10911,9876,10909
-
继续采坑
org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException: sendDefaultImpl call timeout 这个问题是阿里云服务器存在多个网卡,rocketMQ会根据当前网卡选择一个IP使用,我们需要制定一个IP: 路径是: /usr/local/software/rocketmq/distribution/target/apache-rocketmq vim ./conf/broker.conf 添加配置: brokerIP1=公网IP 重新启动: nohup sh bin/mqbroker -n localhost:9876 -c ./conf/broker.conf & tail -f nohup.out- 其他问题
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
消费者操作
-
在前一个项目的基础上,将公共内容提取出来,创建一个JsmConfig的类,来声明公共内容:
public class JmsConfig { /** * 端口 */ public static final String NAME_SERVER = "39.106.214.179:9876"; /** * topic,消息依赖于topic */ public static final String TOPIC = "pay_test_topic"; } -
生产者内容变为
@Component public class PayProducer { /** * 生产组,生产者必须在生产组内 */ private String producerGroup = "pay_producer_group"; private DefaultMQProducer producer; public PayProducer() { producer = new DefaultMQProducer(producerGroup); // 指定nameServer地址,多个地址之间以 ; 隔开 producer.setNamesrvAddr(JmsConfig.NAME_SERVER); start(); } public DefaultMQProducer getProducer() { return producer; } /** * 对象在使用之前必须调用一次,并且只能初始化一次 */ public void start() { try { this.producer.start(); } catch (MQClientException e) { e.printStackTrace(); } } /** * 一般在应用上下文,使用上下文监听器,进行关闭 */ public void shutdown() { producer.shutdown(); } } -
创建消费者
@Component public class PayConsumer { private DefaultMQPushConsumer consumer; private String consumerGroup = "pay_consumer_group"; public PayConsumer() throws MQClientException { consumer = new DefaultMQPushConsumer(consumerGroup); consumer.setNamesrvAddr(JmsConfig.NAME_SERVER); // 设置消费地点,从最后一个进行消费(其实就是消费策略) consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET); // 订阅主题的哪些标签 consumer.subscribe(JmsConfig.TOPIC, "*"); // 注册监听器 consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> { try { // 获取Message 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 Listener"); } } -
Controller的变化:
@RestController public class PayController { @Autowired private PayProducer payProducer; @RequestMapping("/api/v1/pay_cb") public Object callback(String text) throws InterruptedException, RemotingException, MQClientException, MQBrokerException { // 创建消息 主题 二级分类 消息内容好的字节数组 Message message = new Message(JmsConfig.TOPIC, "taga", ("hello rocketMQ " + text).getBytes()); SendResult send = payProducer.getProducer().send(message); System.out.println(send); return new HashMap<>(); } }
梳理一下整个流程,生产者存在于生产组,所以生产组很重要,创建生产者需要指定生产组.消费者同理,创建消费者也需要指定消费组. 并且二者都需要指定NameServer. 有了生产者就要发送消息,也就是Message,创建Message需要指定Topic,二级分类和消息体等信息. 那消费者如何获取呢? 无非就是绑定Topic和二级分类就可以,这就是整个流程. 中间少说了消息的存放,消息是在broker中,这个相当于仓库,所以就是生产者生产消息到Broker,Consumer从Broker中获取消息进行消费.