RocketMQ 消息中间件

集成 SpringBoot
- 入门案例
- 生产消息类型 - 同步、异步、一次性
- 消费模式 - 集群、广播
- 延时消息
- 设置消息标签
- 设置消息的 Key
- 自定义属性设置
- 消息过滤
- 发送消息的方式

Java 从 0 到架构师目录：【Java从0到架构师】学习记录

集成 SpringBoot

入门案例

依赖：

<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.1.0</version>
</dependency>

生产者：

配置文件：

rocketmq.name-server=192.168.52.128:9876
rocketmq.producer.group=my_group
server.port=9999

实现代码：

@RestController
public class HelloController {@Autowiredprivate RocketMQTemplate rocketMQTemplate;@RequestMapping("01_hello")public String sendMsg(String message) throws Exception{SendResult sendResult = rocketMQTemplate.syncSend("01_boot_hello", message);return JSON.toJSONString(sendResult);}
}

消费者：

配置文件：

rocketmq.name-server=192.168.52.128:9876
server.port=7777

实现代码：

@Component
@RocketMQMessageListener(topic = "01_boot_hello",consumerGroup = "maoge_consumer"
)
public class HelloConsumer implements RocketMQListener<String> {@Overridepublic void onMessage(String msg) {System.out.println("接收到的消息:" + msg);}
}

生产消息类型 - 同步、异步、一次性

同步消息：syncSend

rocketMQTemplate.syncSend("01_boot_hello", message);

异步消息：asyncSend

rocketMQTemplate.asyncSend("02_boot_async", message, new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {System.out.println("消息发送成功:" + JSON.toJSONString(sendResult));}@Overridepublic void onException(Throwable throwable) {System.out.println("消息处理失败:" + throwable.getMessage());}
});

一次性消息：sendOneWay

rocketMQTemplate.sendOneWay("03_boot_oneway",message);

消费模式 - 集群、广播

集群模式：

@Component
@RocketMQMessageListener(topic = "02_boot_model",messageModel = MessageModel.CLUSTERING,consumerGroup = "02_boot_cluster"
)
public class ClusterConsumer01 implements RocketMQListener<MessageExt> {@Overridepublic void onMessage(MessageExt message) {System.out.println(JSON.toJSONString(message));}
}

广播模式：

// 广播模式是实时消费消息的，在广播模式消费者启动之前的消息，无法接收
// 广播模式下发送失败的消息不会重试
@Component
@RocketMQMessageListener(topic = "02_boot_model",messageModel = MessageModel.BROADCASTING,consumerGroup = "02_boot_cluster"
)
public class ClusterConsumer02 implements RocketMQListener<MessageExt> {@Overridepublic void onMessage(MessageExt message) {System.out.println(JSON.toJSONString(message));}
}

延时消息

使用原生的 Produce 对象：

DefaultMAProducer producer =  rocketMQTemplate.getProducer();
Message message = new Message(topic, "TagA", "9527", msg.getBytes());
message.setDelayTimeLevel(3);
// 在实际工作中，确保消息可靠性，捕获对应的异常
producer.send(message);

使用 Spring 接口：

Message<String> msg = MessageBuilder.withPayload(message).build();
rocketMQTemplate.syncSend("04_boot_delay", msg, 3000, 3);

设置消息标签

// 在发送的消息 Topic:Tag 中间使用冒号隔开
rocketMQTemplate.convertAndSend("01-boot-hello:TagB", message, map);

设置消息的 Key

Message<?> msg = MessageBuilder.withPayload(message).setHeader(MessageConst.PROPERTY_KEYS, "1100").build();
rocketMQTemplate.send("01-boot-hello", msg);

自定义属性设置

// 过滤设置: 需要开启 broker 的支持用户属性配置
// enablePropertyFilter=trueMap<String,Object> map=new HashMap<>();
//用户自定义属性
map.put("name", "hesj");
map.put("age", "18");
//也可以设置系统属性
map.put(MessageConst.PROPERTY_KEYS,age);
rocketMQTemplate.convertAndSend("01-boot-hello:TagB", message, map);

消息过滤

// 在RocketMQMessageListener添加注解
@RocketMQMessageListener(consumerGroup = "02_boot_cluster",topic = "02_boot_model",messageModel = MessageModel.BROADCASTING,//  消息过滤selectorType = SelectorType.TAG,selectorExpression = "age > 16"
)

发送消息的方式

直接使用 rocketMQTemplate
使用 DefaultMQProducer 对象
使用 Spring 的 Message 接口

【Java从0到架构师】RocketMQ 使用 - 集成 SpringBoot相关推荐

【Java从0到架构师】RocketMQ 使用 - 发送消息、接收消息、特殊消息
RocketMQ 消息中间件入门案例 NameServer 地址发送消息同步发送消息异步发送消息一次性发送消息生产者组.消息封装接收消息消费方式:推式消费.拉式消费消息方式:集群模式 ...
【Java从0到架构师】RocketMQ 基础 - 应用、核心组件、安装
RocketMQ 消息中间件基础知识消息中间件的应用异步解耦削峰填谷消息分发 RocketMQ 核心组件 RocketMQ 安装源码安装修改配置参数启动管理控制台安装测试项目 Ja ...
【Java从0到架构师(1)，Java中高级面试题总结(全面)
JSP 九大内置对象 MySQL 基础 + 多表查询 [Java从0到架构师]MySQL 基础 MySQL MySQL 的使用步骤数据库的内部存储细节 GUI 工具 SQL 语句 DDL 语句 DD ...
【Java从0到架构师】SpringCloud - Sleuth、Zipkin、Config
SpringCloud 链路追踪组件 Sleuth Zipkin 分布式配置中心 - Config Git + Config 分布式配置中心 Java 从 0 到架构师目录:[Java从0到架构师]学 ...
【Java从0到架构师】SpringCloud - Hystrix、Zuul
SpringCloud 基本概念熔断和降级服务雪崩效应服务熔断与降级 - Hystrix SpringBoot 集成 Hystrix 熔断降级服务异常报警通知重点属性 - 熔断隔离策略.超时时 ...
【Java从0到架构师】SpringCloud - Eureka、Ribbon、Feign
SpringCloud 分布式.微服务相关概念微服务框架构选型 SpringCloud 概述服务注册与发现 - Eureka 案例项目 Eureka 自我保护机制微服务调用方式 - Ribbon ...
【Java从0到架构师】Dubbo 基础 - 设置启动时检查、直接提供者、线程模型、负载均衡、集群容错、服务降级
Dubbo 分布式 RPC 分布式核心基础分布式概述 RPC Dubbo Dubbo 入门程序 - XML.注解部署管理控制台 Dubbo Admin 修改绑定的注册 IP 地址设置启动时检查 ...
【Java从0到架构师】分布式框架通信核心基础 - 序列化（JDK、Protobuf）、远程过程调用 RMI
分布式框架通信核心基础序列化 JDK 的序列化 JDK 序列化的一些细节 Protobuf 序列化 Protobuf 环境搭建与操作 Protobuf 原理分析实际数据传输序列化技术选型远程过 ...
【Java从0到架构师】Nginx 拓展 - HTTPS支持、缓存、Http请求防盗链、限流、高可用（Keepalived）
Nginx HTTPS 支持 SSL 证书重写 - rewrite Nginx 跨域解决方案 Nginx 开启缓存 Http 请求防盗链 Nginx 限流方案常见的限流策略 - 计数器算法.漏桶算 ...

【Java从0到架构师】RocketMQ 使用 - 集成 SpringBoot

RocketMQ 消息中间件

集成 SpringBoot

入门案例

生产消息类型 - 同步、异步、一次性

消费模式 - 集群、广播

延时消息

设置消息标签

设置消息的 Key

自定义属性设置

消息过滤

发送消息的方式

【Java从0到架构师】RocketMQ 使用 - 集成 SpringBoot相关推荐

最新文章

热门文章