RabbitMQ(十)：Spring Cloud Stream整合

心已赠人 2023-06-30 06:46 6阅读 0赞

# 1. Spring Cloud Stream应用模型 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5MTk4NzQ5_size_16_color_FFFFFF_t_70]

1.  Spring Cloud Stream应用由第三方的中间件组成。应用间的通信通过输入(input)和输出(output)通道完成，这些通道是由Spring Cloud Stream注入的，而通道与外部的代理（中间件）的连接又是通过Binder实现的；
2.  Binder可以理解为提供了中间件的操作方法的类。Spring Cloud提供了Binder抽象接口以及kafka和rabbitMQ的Binder实现，即封装了对消息系统（kafka，rabbitMQ）的操作。

# 2. 四个重要注解： #

1.  @Output注解: 输出注解，用于定义发送消息接口；
2.  @Input注解: 输入注解，用于定义消息的消费者接口；
3.  @EnableBinding：创建通道，并将通道和Binder绑定。该注解接收的参数就是使用@Input或者@Output注解声明了通道（channel）的接口。Spring Cloud Stream会自动实现这些接口。
4.  @StreamListener: 用于定义监听方法的注解，对消息进行处理。

# 3. 注意 #

1.  使用Spring Cloud Stream是非常简单的，应用好这四个注解即可；
2.  在实现高性能消息的生产和消费的场景非常适合；
3.  存在一个比较严重的问题，就是不能实现消息的可靠性投递，会存在少量消息丢失的问题，除非采用一些补偿措施。

# 4. 具体操作 #

1.  首先引入相关依赖

<dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
                <version>3.0.1.RELEASE</version>
            </dependency>

1.  配置生产者yml文件

spring:
      cloud:
        stream:
          bindings:
            output_channel:
              destination: exchange-3
              group: queue-3
              binder: rabbit #rabbitMQ的Binder的名称就是rabbit
          binders:
            rabbit:
              type: rabbit
              environment:
                spring:
                  rabbitmq:
                    addresses: localhost:5672
                    username: admin
                    password: admin
                    virtual-host: /

1.  创建管道  
    (1). 创建Barista接口

package com.qs.springbootsender.springcloudstream;
    
    import org.springframework.cloud.stream.annotation.Output;
    import org.springframework.messaging.MessageChannel;
    
    /** * 声明通道 * * @author QuS * @date 2020/1/15 15:10 */
    public interface Barista { 
    
        String OUTPUT_CHANNEL = "output_channel";
    	
        @Output(Barista.OUTPUT_CHANNEL)	// 声明通道，此处注意返回值是MessageChannel
        MessageChannel logoutput();
    }

(2). 创建发送端

package com.qs.springbootsender.springcloudstream;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.messaging.Message;
    import org.springframework.messaging.MessageHeaders;
    import org.springframework.messaging.support.MessageBuilder;
    import org.springframework.stereotype.Service;
    
    import java.util.Map;
    
    /** * @author QuS * @date 2020/1/15 15:25 */
    @EnableBinding(Barista.class)//创建Barista定义的通道，并将通道和Binder绑定。
    @Service
    public class RabbitmqSender { 
    
        @Autowired
        private Barista barista;
    
        //发送消息
        public String sendMessage(Object message, Map<String, Object> properties) { 
            try { 
                MessageHeaders mhs = new MessageHeaders(properties);
                Message msg = MessageBuilder.createMessage(message, mhs);
                System.out.println("准备发送>>:" + message);
                //send方法接收一个Message对象，这个对象不能直接new，需要使用MessageBuilder获取。
                boolean sendStatus = barista.logoutput().send(msg);
                System.out.println("发送数据>>:" + message + " ,发送状态>>:" + sendStatus);
            } catch (Exception e) { 
                System.err.println("--------------err---------------");
                e.printStackTrace();
            }
            return null;
        }
    }

1.  接下来是消费者配置，消费端首先引入与生产者端相同依赖spring-cloud-starter-stream-rabbit，然后配置yml文件

spring:
      cloud:
        stream:
          bindings:
            input_channel:
              destination: exchange-3
              group: queue-3
              binder: rabbit
              consumer:
                concurrency: 1
          rabbit:
            bindings:
              input_channel:
                consumer:
                  requeue-rejected: false
                  acknowledge-mode: MANUAL
                  recovery-interval: 3000
                  durable-subscription: true
                  max-concurrency: 5
          binders:
            rabbit:
              type: rabbit
              environment:
                spring:
                  rabbitmq:
                    addresses: localhost:5672
                    username: admin
                    password: admin
                    virtual-host: /

1.  创建消费者端通道

package com.qs.springbootreceiver.springcloudstream;
    
    import org.springframework.cloud.stream.annotation.Input;
    import org.springframework.messaging.SubscribableChannel;
    
    /** * @author QuS * @date 2020/1/15 15:40 */
    public interface Barista { 
    
        String INPUT_CHANNEL = "input_channel";
    
        @Input(Barista.INPUT_CHANNEL) //此处注意返回值是SubscribableChannel 
        SubscribableChannel loginput();
    }

1.  创建消费者端

package com.qs.springbootreceiver.springcloudstream;
    
    import com.rabbitmq.client.Channel;
    import org.springframework.amqp.support.AmqpHeaders;
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.cloud.stream.annotation.StreamListener;
    import org.springframework.messaging.Message;
    import org.springframework.stereotype.Service;
    
    import java.io.IOException;
    
    /** * @author QuS * @date 2020/1/15 15:41 */
    @EnableBinding(Barista.class)
    @Service
    public class RabbitmqReceiver { 
    
        @StreamListener(Barista.INPUT_CHANNEL)
        public void receiver(Message message) throws IOException { 
            Channel channel = (Channel) message.getHeaders().get(AmqpHeaders.CHANNEL);
            Long deliveryTag = (Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
            System.out.println("Input Stream 1 接收数据：" + message);
            System.out.println("消费完毕---------------");
            channel.basicAck(deliveryTag, false);
        }
    }

1.  首先启动消费者端，然后在生产者端编写一个测试，发送一条消息，用来验证！

# 5. 总结 #

整体流程大概如下：

1.  首先打开RabbitMQ服务；
2.  分别编写生产者和消费者的yml配置文件，主要包括destination（rabbitmq对应exchange）、group(rabbitmq对应queue)、所用中间件的地址、端口、用户名、密码等等；
3.  声明通道。通过接口的方式，在接口中使用注解@Input或者@Output去标注；
4.  通过@EnableBinding，创建通道并与Binder绑定；
5.  编写生产者和消费者的主逻辑代码；
6.  进行测试。

***关于更多详情请见：[https://blog.csdn.net/weixin\_38399962/article/details/82192340][https_blog.csdn.net_weixin_38399962_article_details_82192340]***

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5MTk4NzQ5_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200116095903573.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5MTk4NzQ5,size_16,color_FFFFFF,t_70
[https_blog.csdn.net_weixin_38399962_article_details_82192340]: https://blog.csdn.net/weixin_38399962/article/details/82192340