Spring Cloud 终结篇之消息驱动--stream大集合

创建子工程 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

Spring Cloud 终结篇之消息驱动--stream大集合

相关推荐

Spring和SpringBoot及SpringCloud之间的关系是怎样的

java Spring怎么创建Bean的生命周期

Spring Junit测试找不到SpringJUnit4ClassRunner.class怎么解决

Spring整合JUnit的方法是什么

Java之Spring整合Junit的方法是什么

maven error: 用maven下载的spring jar包出现该异常 cannot be read or is not a valid ZIP file

如何进行spring中的默认标签源码解析

Spring中怎么获取其他Bean的属性值