rabbitMQ的简单使用

女爷i 2022-04-05 14:14 311阅读 0赞

### 文章目录 ###

*   *  direct模式：直来直往，指哪到哪儿
     *   *  配置：
         *  生产者：
         *  消费者：
         *  特点：
 *  交换机模式：
 *   *  交换机模式1：topic模式，
     *   *  配置：
         *  生产者：
         *  消费者：
         *  过程：
     *  交换机模式2:Fanout模式：
     *   *  配置：
         *  生产者：
         *  消费者：
         *  特点：
     *  交换机模式3：Headers模式，
     *   *  配置：
         *  生产者：
         *  消费者：
         *  特点：
     *  总结：

生产者：制造消息  
消费者：获取消息，处理消息  
路由机制：中转消息  
四种模式：

后三种是交换机模式

Direct：不需要交换机，直接发送到指定队列，一对一
    Topic：  需要key
    Fanout： 广播模式，不需要key
    Headers：注意模式比较奇特，需要一个map，里面包含需要的header，只有匹配(或者不匹配）其中的header，才能放到对应的队列中

## direct模式：直来直往，指哪到哪儿 ##

代码demo：  
direct模式：

### 配置： ###

@Configuration
    public class MQConfig { 
        public static final String QUEUE="queue";
      /** * 1, diect 模式 * @return */
        @Bean
        public Queue queue(){ 
            return new Queue("queue",true);
        }
    }

### 生产者： ###

@Service
    @Slf4j
    public class MQSender { 
        @Autowired
        AmqpTemplate amqpTemplate;
    
        public void send(Object message){ 
            String msg=RedisService.beanToString(message);
            log.info("send at:{},msg:{} ", MQConfig.QUEUE,msg);
            amqpTemplate.convertAndSend( MQConfig.QUEUE,msg);
        }
    }

### 消费者： ###

@Service
    @Slf4j
    public class MQReceiver { 
        @Autowired
        AmqpTemplate amqpTemplate;
        @RabbitListener(queues=MQConfig.QUEUE)
        public void receive(String message){ 
          log.warn("receive message from {},data:{}",MQConfig.QUEUE,message);
        }
    }

### 特点： ###

很直接，除了生产者和消费者，就没有更多的部分

# 交换机模式： #

比direct多了个中间商（媒婆）,交换机

## 交换机模式1：topic模式， ##

生产者发送消息时需要指定key和交换机，交换机知道key对应的queue，那么交换机就能将消息放到适合的队列中，交换机任务也就完成了。

### 配置： ###

@Configuration
    public class MQConfig { 
    
        public static final String TOPIC_QUEUE1="topic.queue1";
        public static final String TOPIC_QUEUE2="topic.queue2";
      /** * 2 ,topic 模式 交换机 * @return */
        @Bean
        public Queue topicQueue1(){ 
            return new Queue(TOPIC_QUEUE1,true);
        }
        @Bean
        public Queue topicQueue2(){ 
            return new Queue(TOPIC_QUEUE2,true);
        }
        @Bean
        public TopicExchange topicExchange(){ 
            return new TopicExchange(TOPIC_EXCHANGE);
        }
        /** *给队列绑定上key和交换机 */
    
        @Bean
        public Binding topicBinding1(){ 
            return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("topic.key1");
        }
        @Bean
        public Binding topicBinding2(){ 
        /** * # 代表通配符 */
            return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("topic.#");
        }
    }

### 生产者： ###

发送消息时需要指定交换机和key

@Service
    @Slf4j
    public class MQSender { 
        @Autowired
        AmqpTemplate amqpTemplate;
     public void sendTopic(Object message){ 
            String msg=RedisService.beanToString(message);
            log.info("send topic message: {}",msg);
            amqpTemplate.convertAndSend(MQConfig.TOPIC_EXCHANGE,"topic.key1",msg+" 1");
            amqpTemplate.convertAndSend(MQConfig.TOPIC_EXCHANGE,"topic.key2",msg+" 2");
    
        }
    }

### 消费者： ###

只需要监听需要的队列即可

@Service
    @Slf4j
    public class MQReceiver { 
        @Autowired
        AmqpTemplate amqpTemplate;
     /** * topic 交换机模式下的receiver */
        @RabbitListener( queues = MQConfig.TOPIC_QUEUE1)
        public void  receiveTopic1(String message){ 
            log.info("receiveTopic queue1 :{}",message);
        }
        @RabbitListener( queues = MQConfig.TOPIC_QUEUE2)
        public void  receiveTopic2(String message){ 
            log.info("receiveTopic queue2 :{}",message);
        }
    }

### 过程： ###

生产者->message+key1->Exchange->queue(key1)->消费者

## 交换机模式2:Fanout模式： ##

广播模式，不指定key，通过交换机将消息发给与该交换机绑定的队列

### 配置： ###

@Configuration
    public class MQConfig { 
        public static final String QUEUE="queue";
        public static final String TOPIC_QUEUE1="topic.queue1";
        public static final String TOPIC_QUEUE2="topic.queue2";
       
        public static final String FANOUT_EXCHANGE="fanoutExchange";
    /** * 3, Fanout 模式 （广播模式） */
        @Bean
        public FanoutExchange fanoutExchange(){ 
            return new FanoutExchange(FANOUT_EXCHANGE);
        }
        /** * 广播模式绑定 */
        @Bean
        public Binding fanoutBinding1(){ 
            return BindingBuilder.bind(topicQueue1()).to(fanoutExchange());
        }
        @Bean
        public Binding fanoutBinding2(){ 
            return BindingBuilder.bind(topicQueue2()).to(fanoutExchange());
        }
    }

### 生产者： ###

@Service
    @Slf4j
    public class MQSender { 
        @Autowired
        AmqpTemplate amqpTemplate;
      public void sendFanout(Object message){ 
            String msg=RedisService.beanToString(message);
            log.info("send FANOUT message: {}",msg);
            amqpTemplate.convertAndSend(MQConfig.FANOUT_EXCHANGE,"",msg+" FANOUT 1");
    
    
        }
    }

### 消费者： ###

同topic模式下的消费者

### 特点： ###

无需key，发送的是一条消息，但是接受这条小东西的队列一般是多个（广播）

## 交换机模式3：Headers模式， ##

匹配要求很多（通过一个map的key和value逐个与消息发送时设置的properties相比对，有wehereAll，whereAny,where 等等）

### 配置： ###

@Configuration
    public class MQConfig { 
        public  static final String HEADER_QUEUE="header.queue";
        public static final String  HEADER_EXCHANGE="headerExchange";
    /** * 4 header 模式 */
        @Bean
        public HeadersExchange headersExchange(){ 
            return new HeadersExchange(HEADER_EXCHANGE);
        }
        @Bean
        public Queue headerQueue(){ 
            return new Queue(HEADER_QUEUE,true);
        }
        @Bean
        public Binding headerBinding(){ 
            Map<String,Object> map=new HashMap<>();
            map.put("header1","value1");
            map.put("header2","value2");
            //满足符合这个map里面的数据才能放入队列
            return BindingBuilder.bind(headerQueue()).to(headersExchange()).whereAll(map).match();
        }
    }

### 生产者： ###

@Service
    @Slf4j
    public class MQSender { 
        @Autowired
        AmqpTemplate amqpTemplate;
    public void sendHeader(Object message){ 
            String msg=RedisService.beanToString(message);
            log.info("send HEADERS message: {}",msg);
            MessageProperties properties=new MessageProperties();
            properties.setHeader("header1","value1");
            properties.setHeader("header2","value2");
            Message obj=new Message(msg.getBytes(),properties);
            amqpTemplate.convertAndSend(MQConfig.HEADER_EXCHANGE,"",obj);
        }
    }

### 消费者： ###

比较特别，发送和接收都是byte\[\]

@Service
        @Slf4j
        public class MQReceiver { 
            @Autowired
            AmqpTemplate amqpTemplate;
        /** * headers 模式下的receiver */
            @RabbitListener(queues = MQConfig.HEADER_QUEUE)
            public void receiveHeader(byte [] message){ 
                String msg=new String(message);
                log.info("header receiver :{}",msg);
        
            }
        }

### 特点： ###

匹配很灵活，相当于是复合的key；消息是被转为byte\[\]传输和接收

## 总结： ##

direct模式：非交换机模式，直接发送各某个队列

topic模式：交换机+key，交换机通过消息的key去匹配哪个队列可以被放入此消息

fanout模式：交换机+广播 ，不使用key，交换机直接将信息放入与其绑定的队列

headers模式：交换机+map，可以看成是复合的key，交换机去将消息的properties与队列绑定时设置的header map值取匹配