9.介绍Spring Cloud Stream与RabbitMQ集成

逃离我推掉我的手 2023-07-05 05:27 102阅读 0赞

# 介绍Spring Cloud Stream与RabbitMQ集成 #

### 前言 ###

Spring Cloud Stream是一个建立在Spring Boot和Spring Integration之上的框架，有助于创建事件驱动或消息驱动的微服务。在本文中，我们将通过一些简单的例子来介绍Spring Cloud Stream的概念和构造。

*  Bindings — 声明输入和输出通道的接口集合。
 *  Binder — 消息中间件的实现，如Kafka或RabbitMQ
 *  Channel — 表示消息中间件和应用程序之间的通信管道
 *  StreamListeners — bean中的消息处理方法，在中间件的MessageConverter特定事件中进行对象序列化/反序列化之后，将在信道上的消息上自动调用消息处理方法。
 *  Message Schemas — 用于消息的序列化和反序列化，这些模式可以静态读取或者动态加载，支持对象类型的演变。

### **依赖：** ###

compile group: 'org.springframework.cloud', name: 'spring-cloud-starter-stream-rabbit'

#### 生产端： ####

1.yml

spring:
      application:
        name: rabbitmq-springcloudstream-producer
      cloud:
        stream:
          instanceCount: 3
          bindings:
            output_channel:       #输出 生产者
              group: queue-1 #指定相同的exchange-1和不同的queue 表示广播模式 #指定相同的exchange和相同的queue表示集群负载均衡模式
              destination: exchange-1 # kafka:发布订阅模型里面的topic rabbitmq: exchange的概念（但是exchange的类型那里设置呢？）
              binder: local_rabbit
          binders:
            local_rabbit:
              type: rabbit
              environment:
                spring:
                  rabbitmq:
                    host: 10.0.0.22
                    port: 5672
                    username: xiefei
                    password: xiefei
                    virtual-host: /

2.Barista

/** * 这里的Barista接口是定义来作为后面类的参数，这一接口定义来通道类型和通道名称。 * 通道名称是作为配置用，通道类型则决定了app会使用这一通道进行发送消息还是从中接收消息。 * * @author: 谢飞 */
    public interface Barista { 
        String INPUT_CHANNEL = "input_channel";
        String OUTPUT_CHANNEL = "output_channel";
    
    
        //注解@Input声明了它是一个输入类型的通道，名字是Barista.INPUT_CHANNEL，也就是position3的input_channel。这一名字与上述配置app2的配置文件中position1应该一致，表明注入了一个名字叫做input_channel的通道，它的类型是input，订阅的主题是position2处声明的mydest这个主题
        @Input(Barista.INPUT_CHANNEL)
        SubscribableChannel loginput();
    
        //注解@Output声明了它是一个输出类型的通道，名字是output_channel。这一名字与app1中通道名一致，表明注入了一个名字为output_channel的通道，类型是output，发布的主题名为mydest。
        @Output(Barista.OUTPUT_CHANNEL)
        MessageChannel logoutput();
    }

3.RabbitmqSender

/** * @author: 谢飞 */
    @Service
    public class RabbitmqSender { 
    
        @Autowired
        private Barista source;
    
        // 发送消息
        public String sendMessage(Object message){ 
            try{ 
                source.logoutput().send(MessageBuilder.withPayload(message).build());
                System.out.println("发送数据：" + message);
            }catch (Exception e){ 
                e.printStackTrace();
            }
            return null;
        }
    }

4.application

/** * @author 谢飞 */
    @SpringBootApplication
    @EnableBinding(Barista.class)
    public class ProducerApplication { 
    
        public static void main(String[] args) { 
            SpringApplication.run(ProducerApplication.class, args);
        }
    
    }

5.测试

/** * @author: 谢飞 */
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class TestRabbitMQ { 
    
        @Autowired
        private RabbitmqSender rabbitmqSender;
    
        /** * 发送到集群队列 */
        @Test
        public void sendHelloQueue() { 
            rabbitmqSender.sendMessage("这是发送的消息");
        }
    
    }

### 消费端 ###

1.yml

spring:
      application:
        name: rabbitmq-springcloudstream-consumer
      cloud:
        stream:
          instanceCount: 3
          bindings:
            input_channel:       #输出 生产者
              destination: exchange-1 # kafka:发布订阅模型里面的topic rabbitmq: exchange的概念（但是exchange的类型那里设置呢？）
              group: queue-1 #指定相同的exchange-1和不同的queue 表示广播模式 #指定相同的exchange和相同的queue表示集群负载均衡模式
              binder: rabbit_local
              consumer:
                concurrency: 1
          rabbit:
            bindings:
              input_channel:
                consumer:
                  transacted: true
                  txSize: 10
                  acknowledgeMode: MANUAL
                  durableSubscription: true
                  maxConcurrency: 20
                  recoveryInterval: 3000
          binders:
            rabbit_local:
              type: rabbit
              environment:
                spring:
                  rabbitmq:
                    host: 10.0.0.22
                    port: 5672
                    username: xiefei
                    password: xiefei
                    virtual-host: /

2.Barista

3.RabbitmqReceiver

/** * @author: 谢飞 */
    @EnableBinding(Barista.class)
    @Service
    public class RabbitmqReceiver { 
    
        @StreamListener(Barista.INPUT_CHANNEL)
        public void receiver(Message message) { 
            //广播通道
            //PublishSubscribeChannel psc = new PublishSubscribeChannel();
            //确认通道
            //RendezvousChannel rc = new RendezvousChannel();
            Channel channel = (com.rabbitmq.client.Channel) message.getHeaders().get(AmqpHeaders.CHANNEL);
            Long deliveryTag = (Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
            System.out.println("Input Stream 1 接受数据：" + message);
            try { 
                channel.basicAck(deliveryTag, false);
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }
    }

总结：以上就是一个spring Cloud Stream 整合RabbitMQ的一个简单的运用，代码偏多。

代码地址：https://gitee.com/zoo-plus/springboot-learn/tree/2.x/springboot-middleware