RabbitMQ整合Spring Booot【Exchange-Fanout模式】

pom：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

  <modelVersion>4.0.0</modelVersion>

  <groupId>com.toov5.rabibitMQScribe</groupId>

  <artifactId>rabibitMQScribe</artifactId>

  <version>0.0.1-SNAPSHOT</version>

  <dependencies>

        <dependency>

            <groupId>com.rabbitmq</groupId>

            <artifactId>amqp-client</artifactId>

            <version>3.6.5</version>

        </dependency>

    </dependencies>

</project>

　　

连接工具类：

package com.toov5.utils;

import java.io.IOException;

import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.ConnectionFactory;

//没有做成单例的  VirtualHost 需要复用

public class MQConnectionUtils {

    //创建新的连接

    public static Connection newConnection() throws IOException, TimeoutException {

         //创建连接工厂

    ConnectionFactory factory= new ConnectionFactory();

    //链接地址

    factory.setHost("192.168.91.6");

    //用户名称

    factory.setUsername("admin");

    //用户密码

    factory.setPassword("admin");

    //amqp端口号

    factory.setPort(5672);

    //连接virtualhost

    factory.setVirtualHost("/admin_toov5");

    Connection connection = factory.newConnection();

        return connection;

    }

}

生产者：

package com.toov5.fanout;

import java.io.IOException;

import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.toov5.utils.MQConnectionUtils;

//生产者 交换机类型 producerFanout类型

public class FanoutProducer {

    //交换机名称

     private static final String EXCHANGE_NAME = "my_fanout";

     public static void main(String[] args) throws IOException, TimeoutException {

        //建立MQ连接

         Connection connection = MQConnectionUtils.newConnection();

        //创建通道

          Channel channel = connection.createChannel();

          //生产者绑定交换机

          channel.exchangeDeclare(EXCHANGE_NAME, "fanout");  //交换机名称  交换机类型

          //创建对应的消息

          String msString = "my_fanout_destination_msg";

          //通过频道 发送消息

          System.out.println("生产者投递消息："+msString);

          //消息投递到交换机里面去

          channel.basicPublish(EXCHANGE_NAME, "", null, msString.getBytes());

          //关闭通道 和 连接

          channel.close();

          connection.close();

    }

}

消费者

package com.toov5.fanout;

import java.io.IOException;

import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.DefaultConsumer;

import com.rabbitmq.client.Envelope;

import com.rabbitmq.client.AMQP.BasicProperties;

import com.toov5.utils.MQConnectionUtils;

//邮件消费者

public class ConsumerEmailFanout {

    private static final String EMAIL_QUEUE ="email_queue_fanout";

    //交换机名称

   private static final String EXCHANGE_NAME = "my_fanout";

     public static void main(String[] args) throws IOException, TimeoutException {

         System.out.println("邮件消费者启动");

        //建立MQ连接

         Connection connection = MQConnectionUtils.newConnection();

        //创建通道

          Channel channel = connection.createChannel();

       //消费者声明队列

          channel.queueDeclare(EMAIL_QUEUE, false, false, false, null);

         //消费者队列绑定交换机

          channel.queueBind(EMAIL_QUEUE, EXCHANGE_NAME, "");

          //消费者监听消息

      DefaultConsumer defaultConsumer =     new DefaultConsumer(channel) {

              //重写监听方法

             @Override

             public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)

                     throws IOException {

                 String msg = new String(body,"UTF-8");

                 System.out.println("邮件消费者获取生产者消息"+msg);

             }

         };

         channel.basicConsume(EMAIL_QUEUE,true, defaultConsumer);   //绑定队列 事件监听

    }

}

package com.toov5.fanout;

import java.io.IOException;

import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.DefaultConsumer;

import com.rabbitmq.client.Envelope;

import com.rabbitmq.client.AMQP.BasicProperties;

import com.toov5.utils.MQConnectionUtils;

//邮件消费者

public class ConsumerSMSFanout {

    private static final String SMS_QUEUE ="sms_queue_fanout";

    //交换机名称

   private static final String EXCHANGE_NAME = "my_fanout";

     public static void main(String[] args) throws IOException, TimeoutException {

         System.out.println("短信消费者启动");

        //建立MQ连接

         Connection connection = MQConnectionUtils.newConnection();

        //创建通道

          Channel channel = connection.createChannel();

       //消费者声明队列

          channel.queueDeclare(SMS_QUEUE, false, false, false, null);

         //消费者队列绑定交换机

          channel.queueBind(SMS_QUEUE, EXCHANGE_NAME, "");

          //消费者监听消息

      DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {

              //重写监听方法

             @Override

             public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)

                     throws IOException {

                 String msg = new String(body,"UTF-8");

                 System.out.println("邮件消费者获取生产者消息"+msg);

             }

         };

         channel.basicConsume(SMS_QUEUE,true, defaultConsumer);   //绑定队列 事件监听

    }

}