zoukankan      html  css  js  c++  java
  • RocketMQ 自定义 Binding

    概述
    在实际生产中,我们需要发布和订阅的消息可能不止一种 Topic ,故此时就需要使用自定义 Binding 来帮我们实现多 Topic 的发布和订阅功能

    生产者

    自定义 Output 接口,代码如下:

    public interface MySource {
    @Output("output1")
    MessageChannel output1();

    @Output("output2")
    MessageChannel output2();
    

    }
    发布消息的案例代码如下:

    @Autowired
    private MySource source;

    public void send(String msg) throws Exception {
    source.output1().send(MessageBuilder.withPayload(msg).build());
    }

    消费者

    自定义 Input 接口,代码如下:

    public interface MySink {
    @Input("input1")
    SubscribableChannel input1();

    @Input("input2")
    SubscribableChannel input2();
    
    @Input("input3")
    SubscribableChannel input3();
    
    @Input("input4")
    SubscribableChannel input4();
    

    }
    接收消息的案例代码如下:

    @StreamListener("input1")
    public void receiveInput1(String receiveMsg) {
    System.out.println("input1 receive: " + receiveMsg);
    }

    Application

    配置 Input 和 Output 的 Binding 信息并配合 @EnableBinding 注解使其生效,代码如下:

    @SpringBootApplication
    @EnableBinding({ MySource.class, MySink.class })
    public class RocketMQApplication {
    public static void main(String[] args) {
    SpringApplication.run(RocketMQApplication.class, args);
    }
    }

    application.yml

    生产者

    spring:
    application:
    name: rocketmq-provider
    cloud:
    stream:
    rocketmq:
    binder:
    namesrv-addr: 192.168.10.149:9876
    bindings:
    output1: {destination: test-topic1, content-type: application/json}
    output2: {destination: test-topic2, content-type: application/json}

    消费者

    spring:
    application:
    name: rocketmq-consumer
    cloud:
    stream:
    rocketmq:
    binder:
    namesrv-addr: 192.168.10.149:9876
    bindings:
    input: {consumer.orderly: true}
    bindings:
    input1: {destination: test-topic1, content-type: text/plain, group: test-group, consumer.maxAttempts: 1}
    input2: {destination: test-topic1, content-type: text/plain, group: test-group, consumer.maxAttempts: 1}
    input3: {destination: test-topic2, content-type: text/plain, group: test-group, consumer.maxAttempts: 1}
    input4: {destination: test-topic2, content-type: text/plain, group: test-group, consumer.maxAttempts: 1

    等你看到的时候,想变得有一点点不一样
  • 相关阅读:
    4.FFMPEGAVFrame
    从0到1实现Web端H.265播放器:视频解码篇
    6.AVCodecContext和AVCodec
    7.SwrContext音频重采样使用
    5.AVStream和AVCodecParameters
    3.AVPacket使用
    FFmpeg:AVFrame结构体分析
    9.下载ffmpeg、使QT支持同时编译32位和64位
    1.ffmpeg基础经常使用知识
    2.AVFormatContext和AVInputFormat
  • 原文地址:https://www.cnblogs.com/snake107/p/11920902.html
Copyright © 2011-2022 走看看