013 RabbitMQ消息发送

ゝ一纸荒年。 2022-03-09 10:58 238阅读 0赞

#### 1.发送 ####

@Component
    public class MqSenderTest extends OrderApplicationTests {
    
        @Autowired
        private AmqpTemplate amqpTemplate;
    
        @Test
        public  void send(){
            amqpTemplate.convertAndSend("myQueue","now:"+new Date());
        }
    
    
        @Test
        public  void sendOrder(){
            amqpTemplate.convertAndSend("myOrder","computer","now:"+new Date());
        }
    }

#### 2.接收 ####

@Slf4j
    @Component
    public class MqReceiver {
    
    
    
        //1.@RabbitListener(queues = "myQueue")
        //2.自动创建队列 @RabbitListener(queuesToDeclare = @Queue("myQueue"))
        //3.自动创建，Exhange和Queue绑定
        @RabbitListener(bindings = @QueueBinding(
                value = @Queue("myQueue"),
                exchange = @Exchange("myExchange")
        ))
        public  void process(String message){
            log.info("************************* MqReceiver:{}",message);
        }
    
    
        @RabbitListener(bindings = @QueueBinding(
                exchange = @Exchange("myOrder"),
                key = "computer",
                value = @Queue("computerOrder")
        ))
        public  void processComputer(String message){
            log.info("************************* processComputer MqReceiver:{}",message);
        }
    
        @RabbitListener(bindings = @QueueBinding(
                exchange = @Exchange("myOrder"),
                key = "fruit",
                value = @Queue("fruitOrder")
        ))
        public  void processFruit(String message){
            log.info("************************* processFruit MqReceiver:{}",message);
        }
    
    }

## 二、使用rabbit stream ##

#### 1.添加依赖 ####

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

#### 2.定义Input Output ####

import org.springframework.cloud.stream.annotation.Input;
    import org.springframework.cloud.stream.annotation.Output;
    import org.springframework.messaging.MessageChannel;
    import org.springframework.messaging.SubscribableChannel;
    
    public interface StreamClient {
    
        String INPUT="myMessageInput";
    
        String OUTPUT="myMessageOutput";
    
        @Input(StreamClient.INPUT)
        SubscribableChannel input();
    
    
        @Output(StreamClient.OUTPUT)
        MessageChannel output();
    }

#### 3.定义接收 ####

import lombok.extern.slf4j.Slf4j;
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.cloud.stream.annotation.StreamListener;
    import org.springframework.messaging.handler.annotation.SendTo;
    import org.springframework.stereotype.Component;
    
    @Component
    @EnableBinding(StreamClient.class)
    @Slf4j
    public class StreamReceiver {
    
        @StreamListener(StreamClient.INPUT)
        @SendTo(StreamClient.OUTPUT)
        public Object processInput(Object message){
            log.info("StreamReceiver:{}",message);
            return  message;
        }
    
       @StreamListener(StreamClient.OUTPUT)
        public  void process(String message){
            log.info("**************************** StreamReceiver:{}",message);
        }
    }

#### 4.myMessageInput 分组，避免同一类型服务多个启动时发送多次 ####

spring:
      application:
        name: order
      cloud:
        config:
          discovery:
            enabled: true
            service-id: CONFIG
          profile: test
        stream:
          binders:
            myMessageInput:
              group: order
              content-type: application/json  #使用json格式传递和接收

#### 5.发送请求 ####

@RestController
    public class SendMessageController{
    
        @Autowired
        private StreamClient streamClient;
    
    //    @GetMapping("/sendMessage")
    //    public void process(){
    //        String message ="now "+ new Date();
    //        streamClient.output().send(MessageBuilder.withPayload(message).build());
    //    }
    
        /**
         * 发送 orderDTO
         */
        @GetMapping("/sendMessage")
        public void process(){
            OrderDTO orderDTO = new OrderDTO();
            orderDTO.setOrderId("123456");
            streamClient.input().send(MessageBuilder.withPayload(orderDTO).build());
        }
    }

#### 发送成功后，发送成功的消息@SendTo(StreamClient.INPUT2) ####

@StreamListener(StreamClient.INPUT)
    @SendTo(StreamClient.INPUT2)
    public  String process(OrderDTO message){
        log.info("**************************** StreamReceiver:{}",message);
        return "receiverd.";
    }
    
    @StreamListener(StreamClient.INPUT2)
    public  void process2(String message){
        log.info("**************************** StreamReceiver:{}",message);
    }