springboot整合rabbitmq（fanout广播模式）

生产者

• 新建producer（springboot-order-rabbitmq-producer）
  
• 修改application.yml

server:
  port: 8080

spring:
  rabbitmq:
    username: admin
    password: admin
    virtual-host: /
    host: 47.100.41.55
    port: 5672

• 模拟一个生产订单服务

@Service
public class OrderService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 
     * @param userId
     * @param productId
     * @param num
     */
    public void makeOrder(String userId,String productId,int num){

        String orderId = UUID.randomUUID().toString();
        System.out.println("订单生成成功：" + orderId);

        String exchangeName = "fanout_order_exchange";
        String routingKey = "";
        //@param1 交换机 @param2 路由key/queue队列名称 @param3 消息内容
        rabbitTemplate.convertAndSend(exchangeName,routingKey,orderId);
    }
}

• 创建交换机，队列的关系

@Configuration
public class RabbitMqConfiguration {


    //声明交换机
    @Bean
    public FanoutExchange fanoutExchange(){

        return new FanoutExchange("fanout_order_exchange",true,false);
    }

    //声明队列
    @Bean
    public Queue smsQueue(){
        return new Queue("sms.fanout.queue",true);
    }
    @Bean
    public Queue duanxinQueue(){
        return new Queue("duanxin.fanout.queue",true);
    }
    @Bean
    public Queue emailQueue(){
        return new Queue("email.fanout.queue",true);
    }

    //完成交换机和队列绑定
    @Bean
    public Binding smsBinding(){
        return BindingBuilder.bind(smsQueue()).to(fanoutExchange());
    }
    @Bean
    public Binding duanxinBinding(){
        return BindingBuilder.bind(duanxinQueue()).to(fanoutExchange());
    }
    @Bean
    public Binding emailBinding(){
        return BindingBuilder.bind(emailQueue()).to(fanoutExchange());
    }
}

• 测试生产者运行

@SpringBootTest
class SpringbootOrderRabbitmqProducerApplicationTests {
    @Autowired
    private OrderService orderService;
    @Test
    void contextLoads() {
        orderService.makeOrder("1","1",12);
    }
}

同时在浏览器打开rabbitmq ​​http://47.100.41.55:15672/​​ 切换到exchanges，queues

消费者

• 新建consumer（springboot-order-rabbitmq-consumer）同上
• 修改application.yml，其他一样，端口号改成8081
• 建立对应的consumer类

@Service
@RabbitListener(queues = {"duanxin.fanout.queue"})
public class FanoutDuanxinConsumer {

    @RabbitHandler
    public void receiveMsg(String msg){
        System.out.println("FanoutDuanxinConsumer ---接收到的订单信息是：->" + msg);
    }
}

@Service
@RabbitListener(queues = {"email.fanout.queue"})
public class FanoutEmailConsumer {

    @RabbitHandler
    public void receiveMsg(String msg){
        System.out.println("FanoutEmailConsumer ---接收到的订单信息是：->" + msg);
    }
}

@Service
@RabbitListener(queues = {"sms.fanout.queue"})
public class FanoutSmsConsumer {

    @RabbitHandler
    public void receiveMsg(String msg){
        System.out.println("FanoutSmsConsumer ---接收到的订单信息是：->" + msg);
    }
}

• 启动SpringbootOrderRabbitmqConsumerApplication
• 查看控制台
  这是刚才生产者发送的消息
  
  生产者再次发送一条消息，消费者依然可以收到