RobbotMQ--整合SpringCloudStream--学习笔记

柔光的暖阳◎ 2023-07-11 12:57 43阅读 0赞

**Barista接口**：Barista接口是定义来作为后面类的参数，这一接口定义来通道类型和通道名称，通道配置作为配置用。通道类型决定了app会使用这一通道进行发送消息还是接收消息。

*  @Output: 输出注解，用于定义发送消息接口
 *  @Input: 输入注解，用于定义消费者接口
 *  @StreamListener: 用于定义监听方法的注解
 *  不能实现可靠性投递，无法保证100%可靠性，存在少量信息丢失问题。（适合kafka）

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

这是个公共类

public interface Barista {
    
        String OUTPUT_CHANNEL = "output_channel";
        String INPUT_CHANNEL = "input_channel";
    
        @Output(Barista.OUTPUT_CHANNEL)
        MessageChannel logoutput();
    
        @Input(Barista.INPUT_CHANNEL)
        SubscribableChannel loginput();
    }

## product ##

**properties**

server.port=8001
    server.servlet.context-path=/producer
    
    spring.application.name=producer
    spring.cloud.stream.bindings.output_channel.destination=exchange-3  # 设置发送到的交换机
    spring.cloud.stream.bindings.output_channel.group=queue-3 # 管道名称
    spring.cloud.stream.bindings.output_channel.binder=rabbit_cluster  # 转发规则
    
    spring.cloud.stream.binders.rabbit_cluster.type=rabbit
    spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.addresses=192.168.11.76:5672
    spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.username=guest
    spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.password=guest
    spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.virtual-host=/

@EnableBinding(Barista.class)
    @Service  
    public class RabbitmqSender {  
      
        @Autowired  
        private Barista barista;  
        
        // 发送消息
        public String sendMessage(Object message, Map<String, Object> properties) throws Exception {  
            try{
            	MessageHeaders mhs = new MessageHeaders(properties);
            	Message msg = MessageBuilder.createMessage(message, mhs);
                boolean sendStatus = barista.logoutput().send(msg);
                System.err.println("--------------sending -------------------");
                System.out.println("发送数据：" + message + ",sendStatus: " + sendStatus);
            }catch (Exception e){  
            	System.err.println("-------------error-------------");
            	e.printStackTrace();
                throw new RuntimeException(e.getMessage());
            }  
            return null;
        }  
        
    }

## **consumer** ##

**properties **

server.port=8002
    server.context-path=/consumer
    
    spring.application.name=consumer
    spring.cloud.stream.bindings.input_channel.destination=exchange-3  # 交换机
    spring.cloud.stream.bindings.input_channel.group=queue-3 # 管道
    spring.cloud.stream.bindings.input_channel.binder=rabbit_cluster # 路由规则
    spring.cloud.stream.bindings.input_channel.consumer.concurrency=1 
    spring.cloud.stream.rabbit.bindings.input_channel.consumer.requeue-rejected=false # 消息处理失败后直接删除，true的话会重新投放
    spring.cloud.stream.rabbit.bindings.input_channel.consumer.acknowledge-mode=MANUAL #手动ack
    spring.cloud.stream.rabbit.bindings.input_channel.consumer.recovery-interval=3000 # 服务断开,3秒钟重连
    spring.cloud.stream.rabbit.bindings.input_channel.consumer.durable-subscription=true
    spring.cloud.stream.rabbit.bindings.input_channel.consumer.max-concurrency=5
    
    # 使用的环境是rabbit
    spring.cloud.stream.binders.rabbit_cluster.type=rabbit
    spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.addresses=192.168.11.76:5672
    spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.username=guest
    spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.password=guest
    spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.virtual-host=/

@EnableBinding(Barista.class)
    @Service
    public class RabbitmqReceiver {  
    
        @StreamListener(Barista.INPUT_CHANNEL)  
        public void receiver(Message message) throws Exception {  
    		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);
        	System.out.println("消费完毕------------");
        	channel.basicAck(deliveryTag, false);
        }  
    }

consummer 配置解释

1 # 端口号
     2 server.port=8002
     3 # 访问路径
     4 server.context-path=/consumer
     5 
     6 # 应用的名称
     7 spring.application.name=consumer
     8 # 交换机名称
     9 spring.cloud.stream.bindings.input_channel.destination=exchange-3
    10 # 队列名称
    11 spring.cloud.stream.bindings.input_channel.group=queue-3
    12 # 对集群环境进行配置
    13 spring.cloud.stream.bindings.input_channel.binder=rabbit_cluster
    14 # 设置默认监听数
    15 spring.cloud.stream.bindings.input_channel.consumer.concurrency=1
    16 # 是否支持重新放回队列
    17 spring.cloud.stream.rabbit.bindings.input_channel.consumer.requeue-rejected=false
    18 # 接收模式是手工接收
    19 spring.cloud.stream.rabbit.bindings.input_channel.consumer.acknowledge-mode=MANUAL
    20 # 服务断开,3秒钟重连
    21 spring.cloud.stream.rabbit.bindings.input_channel.consumer.recovery-interval=3000
    22 # 是否启用持久化订阅
    23 spring.cloud.stream.rabbit.bindings.input_channel.consumer.durable-subscription=true
    24 # 最大的监听数
    25 spring.cloud.stream.rabbit.bindings.input_channel.consumer.max-concurrency=5
    26 
    27 # rabbit_cluster对应上面的spring.cloud.stream.bindings.output_channel.binder的值。名称可以自定义
    28 # 使用的环境是rabbit
    29 spring.cloud.stream.binders.rabbit_cluster.type=rabbit
    30 # 访问地址和端口号
    31 spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.addresses=192.168.110.133:5672
    32 # 账号
    33 spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.username=guest
    34 # 密码
    35 spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.password=guest
    36 # 虚拟主机
    37 spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.virtual-host=/