Spring Cloud 终结篇之消息驱动--stream大集合
程序员文章站
2022-06-19 08:45:00
创建子工程 stream-sample编写pom文件 org.springframework.boot spring-boot-starter-web ....
创建子工程 stream-sample
编写pom文件
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
</dependencies>
创建启动引导类 StreamApplication
@SpringBootApplication
public class StreamApplication {
public static void main(String[] args) {
SpringApplication.run(StreamApplication.class, args);
}
}
创建配置文件
spring.application.name=stream-sample
server.port=63003
# RabbitMQ连接字符串
spring.rabbitmq.host=192.168.0.201
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
# 绑定Channel到broadcast
spring.cloud.stream.bindings.myTopic-consumer.destination=broadcast
spring.cloud.stream.bindings.myTopic-producer.destination=broadcast
# 消息分组示例
spring.cloud.stream.bindings.group-consumer.destination=group-topic
spring.cloud.stream.bindings.group-producer.destination=group-topic
spring.cloud.stream.bindings.group-consumer.group=Group-A
## 消息分区配置
## 打开消费者的消费分区功能
spring.cloud.stream.bindings.group-consumer.consumer.partitioned=true
## 两个消息分区
spring.cloud.stream.bindings.group-producer.producer.partition-count=2
# SpEL (Key resolver) 可以定义复杂表达式生成Key
# 我们这里用最简化的配置,只有索引参数为1的节点(消费者),才能消费消息
spring.cloud.stream.bindings.group-producer.producer.partition-key-expression=1
# 当前消费者实例总数
spring.cloud.stream.instanceCount=2
# 最大值instanceCount-1,当前实例的索引号
spring.cloud.stream.instanceIndex=1
# 延迟消息配置
spring.cloud.stream.bindings.delayed-consumer.destination=delayed-topic
spring.cloud.stream.bindings.delayed-producer.destination=delayed-topic
spring.cloud.stream.rabbit.bindings.delayed-producer.producer.delayed-exchange=true
# 异常消息(单机版重试)
spring.cloud.stream.bindings.error-consumer.destination=error-out-topic
spring.cloud.stream.bindings.error-producer.destination=error-out-topic
# 重试次数(本机重试)
# 次数=1相当于不重试
spring.cloud.stream.bindings.error-consumer.consumer.max-attempts=2
# 异常消息(requeue重试)
spring.cloud.stream.bindings.requeue-consumer.destination=requeue-topic
spring.cloud.stream.bindings.requeue-producer.destination=requeue-topic
# 必须把max-attempts设置为1,否则requeue不能生效
spring.cloud.stream.bindings.requeue-consumer.consumer.max-attempts=1
spring.cloud.stream.bindings.requeue-consumer.group=requeue-group
# 仅对当前requeue-consumer,开启requeue
spring.cloud.stream.rabbit.bindings.requeue-consumer.consumer.requeueRejected=true
# 默认全局开启requeue
# spring.rabbitmq.listener.default-requeue-rejected=true
# 死信队列配置
spring.cloud.stream.bindings.dlq-consumer.destination=dlq-topic
spring.cloud.stream.bindings.dlq-producer.destination=dlq-topic
spring.cloud.stream.bindings.dlq-consumer.consumer.max-attempts=2
spring.cloud.stream.bindings.dlq-consumer.group=dlq-group
# 开启死信队列(默认 topic.dlq)
spring.cloud.stream.rabbit.bindings.dlq-consumer.consumer.auto-bind-dlq=true
# Fallback配置
spring.cloud.stream.bindings.fallback-consumer.destination=fallback-topic
spring.cloud.stream.bindings.fallback-producer.destination=fallback-topic
spring.cloud.stream.bindings.fallback-consumer.consumer.max-attempts=2
spring.cloud.stream.bindings.fallback-consumer.group=fallback-group
# input channel -> fallback-topic.fallback-group.errors
management.security.enabled=false
management.endpoints.web.exposure.include=*
management.endpoint.health.show-details=always
创建 Topic
延时消息
public interface DelayedTopic {
String INPUT = "delayed-consumer";
String OUTPUT = "delayed-producer";
@Input(INPUT)
SubscribableChannel input();
@Output(OUTPUT)
MessageChannel output();
}
死信队列
public interface DlqTopic {
String INPUT = "dlq-consumer";
String OUTPUT = "dlq-producer";
@Input(INPUT)
SubscribableChannel input();
@Output(OUTPUT)
MessageChannel output();
}
异常消息
public interface ErrorTopic {
String INPUT = "error-consumer";
String OUTPUT = "error-producer";
@Input(INPUT)
SubscribableChannel input();
@Output(OUTPUT)
MessageChannel output();
}
fallback降级
public interface FallbackTopic {
String INPUT = "fallback-consumer";
String OUTPUT = "fallback-producer";
@Input(INPUT)
SubscribableChannel input();
@Output(OUTPUT)
MessageChannel output();
}
分区分组
public interface GroupTopic {
String INPUT = "group-consumer";
String OUTPUT = "group-producer";
@Input(INPUT)
SubscribableChannel input();
@Output(OUTPUT)
MessageChannel output();
}
重入队列
public interface RequeueTopic {
String INPUT = "requeue-consumer";
String OUTPUT = "requeue-producer";
@Input(INPUT)
SubscribableChannel input();
@Output(OUTPUT)
MessageChannel output();
}
我的 消息
public interface MyTopic {
String INPUT = "myTopic-consumer";
String OUTPUT = "myTopic-producer";
@Input(INPUT)
SubscribableChannel input();
@Output(OUTPUT)
MessageChannel output();
}
创建stream 流 消息消费者
@Slf4j
@EnableBinding(value = {
Sink.class,
MyTopic.class,
GroupTopic.class,
DelayedTopic.class,
ErrorTopic.class,
RequeueTopic.class,
DlqTopic.class,
FallbackTopic.class
}
)
public class StreamConsumer {
private AtomicInteger count = new AtomicInteger(1);
@StreamListener(Sink.INPUT)
public void consume(Object payload) {
log.info("message consumed successfully, payload={}", payload);
}
// 自定义消息广播
@StreamListener(MyTopic.INPUT)
public void consumeMyMessage(Object payload) {
log.info("My message consumed successfully, payload={}", payload);
}
// 消息分组 & 消费分区示例
@StreamListener(GroupTopic.INPUT)
public void consumeGroupMessage(Object payload) {
log.info("Group message consumed successfully, payload={}", payload);
}
// 延迟消息示例
@StreamListener(DelayedTopic.INPUT)
public void consumeDelayedMessage(MessageBean bean) {
log.info("Delayed message consumed successfully, payload={}", bean.getPayload());
}
// 异常重试(单机版)
@StreamListener(ErrorTopic.INPUT)
public void consumeErrorMessage(MessageBean bean) {
log.info("Are you OK?");
if (count.incrementAndGet() % 3 == 0) {
log.info("Fine, thank you. And you?");
count.set(0);
} else {
log.info("What's your problem?");
throw new RuntimeException("I'm not OK");
}
}
// 异常重试(联机版-重新入列)
@StreamListener(RequeueTopic.INPUT)
public void requeueErrorMessage(MessageBean bean) {
log.info("Are you OK?");
try {
Thread.sleep(3000L);
} catch (Exception e) {
}
// throw new RuntimeException("I'm not OK");
}
// 死信队列
@StreamListener(DlqTopic.INPUT)
public void consumeDlqMessage(MessageBean bean) {
log.info("Dlq - Are you OK?");
if (count.incrementAndGet() % 3 == 0) {
log.info("Dlq - Fine, thank you. And you?");
} else {
log.info("Dlq - What's your problem?");
throw new RuntimeException("I'm not OK");
}
}
// Fallback + 升级版本
@StreamListener(FallbackTopic.INPUT)
public void goodbyeBadGuy(MessageBean bean,
@Header("version") String version) {
log.info("Fallback - Are you OK?");
if ("1.0".equalsIgnoreCase(version)) {
log.info("Fallback - Fine, thank you. And you?");
} else if ("2.0".equalsIgnoreCase(version)) {
log.info("unsupported version");
throw new RuntimeException("I'm not OK");
} else {
log.info("Fallback - version={}", version);
}
}
// 降级流程
@ServiceActivator(inputChannel = "fallback-topic.fallback-group.errors")
public void fallback(Message<?> message) {
log.info("fallback entered");
}
}
创建一个 messageBean
@Data
public class MessageBean {
private String payload;
}
最后一步 创建 Controller
@RestController
@Slf4j
public class Controller {
@Autowired
private MyTopic producer;
@Autowired
private GroupTopic groupTopicProducer;
@Autowired
private DelayedTopic delayedTopicProducer;
@Autowired
private ErrorTopic errorTopicProducer;
@Autowired
private RequeueTopic requeueTopicProducer;
@Autowired
private DlqTopic dlqTopicProducer;
@Autowired
private FallbackTopic fallbackTopicProducer;
// 简单广播消息
@PostMapping("send")
public void sendMessage(@RequestParam(value = "body") String body) {
producer.output().send(MessageBuilder.withPayload(body).build());
}
// 消息分组和消息分区
@PostMapping("sendToGroup")
public void sendMessageToGroup(@RequestParam(value = "body") String body) {
groupTopicProducer.output().send(MessageBuilder.withPayload(body).build());
}
// 延迟消息
@PostMapping("sendDM")
public void sendDelayedMessage(
@RequestParam(value = "body") String body,
@RequestParam(value = "seconds") Integer seconds) {
MessageBean msg = new MessageBean();
msg.setPayload(body);
log.info("ready to send delayed message");
delayedTopicProducer.output().send(
MessageBuilder.withPayload(msg)
.setHeader("x-delay", seconds * 1000)
.build());
}
// 异常重试(单机版)
@PostMapping("sendError")
public void sendErrorMessage(@RequestParam(value = "body") String body) {
MessageBean msg = new MessageBean();
msg.setPayload(body);
errorTopicProducer.output().send(MessageBuilder.withPayload(msg).build());
}
// 异常重试(联机版 - 重新入列)
@PostMapping("requeue")
public void sendErrorMessageToMQ(@RequestParam(value = "body") String body) {
MessageBean msg = new MessageBean();
msg.setPayload(body);
requeueTopicProducer.output().send(MessageBuilder.withPayload(msg).build());
}
// 死信队列测试
@PostMapping("dlq")
public void sendMessageToDlq(@RequestParam(value = "body") String body) {
MessageBean msg = new MessageBean();
msg.setPayload(body);
dlqTopicProducer.output().send(MessageBuilder.withPayload(msg).build());
}
// fallback + 升版
@PostMapping("fallback")
public void sendMessageToFallback(
@RequestParam(value = "body") String body,
@RequestParam(value = "version", defaultValue = "1.0") String version) {
MessageBean msg = new MessageBean();
msg.setPayload(body);
fallbackTopicProducer.output().send(
MessageBuilder.withPayload(msg)
.setHeader("version", version)
.build());
}
}
附:
1. 下载插件
https://www.rabbitmq.com/community-plugins.html
找到rabbitmq_delayed_message_exchange
下载对应版本的插件,3.6和3.7版本插件不一样
2. 下载以后解压,copy到rabbitmq安装目录下的plugins文件夹
3. 安装插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
4. 安装完一定要重启RabbitMQ,不是单单重启UI管理界面!
如果只是单单调用rabbitmqctl stop_app然后再rabbitmqctl start_app是没有作用的!
正确的步骤是先rabbitmqctl stop,然后再直接执行rabbitmq-server
如果以上步骤还能使延迟队列生效,在重启完之后,换一个新的topic名字就好了
本文地址:https://blog.csdn.net/weixin_38305866/article/details/109941990