rabbitMq-发布订阅模式fanout-java


工作原理:生产者经过交换机向队列中发送消息，交换机会把消息发送到订阅了当前消息的队列中。
交换机实现的是动态的数据分发。分发的原则是谁订阅了我，我向谁发。
C1只能消费第一个对列中的内容。
C2只能消费第二个对列中的内容。
发布订阅模式只能做到单纯地进行发布，不能指定某一个队列进行发布。

[root@bogon rabbitmq-server-3.6.1]# cd /etc/rabbitmq

[root@bogon rabbitmq]# service rabbitmq-server start
Starting rabbitmq-server: SUCCESS
rabbitmq-server.er.

pom.xml

<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.rabbit</groupId>
    <artifactId>schoolmanage</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <packaging>war</packaging>
    <dependencies>
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>3.5.1</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit</artifactId>
            <version>1.4.0.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.11</version>
        </dependency>
    </dependencies>
</project>

package schoolmanage;

import java.io.IOException;

import org.junit.Before;
import org.junit.Test;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

public class TestStudentMsgFanOut {
    private Connection connection = null;

    @Before
    public void init() throws IOException {
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 为工厂对象添加数据
        // 远程主机
        factory.setHost("192.168.6.130");
        // 端口号
        factory.setPort(5672);
        // 虚拟主机
        factory.setVirtualHost("/school");
        // 用户名
        factory.setUsername("student");
        // 密码
        factory.setPassword("student");
        // 创建连接
        connection = factory.newConnection();
    }

    // 消息生产者
    @Test
    public void provider() throws IOException {
        // 创建通道
        Channel channel = connection.createChannel();

        // 创建交换机
        String exchangeName = "E1";
        /**
         * 定义交换机模式 String exchange String type fanout发布订阅 redirect路由模式 topic主题模式
         */
        channel.exchangeDeclare(exchangeName, "fanout");

        for (int i = 0; i < 6; i++) {
            String msg = "今天学校下雨了!!!——发布订阅模式,这是第" + i + "次通知";
            channel.basicPublish(exchangeName, "", null, msg.getBytes());
        }
        // 将流关闭
        channel.close();
        connection.close();
        System.out.println("消息发送成功!!!");
    }

    // 消费者1
    @Test
    public void consumer1() throws Exception {
        // 创建通道
        Channel channel = connection.createChannel();

        // 定义对列
        String queueName = "counsumer1";
        channel.queueDeclare(queueName, false, false, false, null);
        // 订阅E1交换机的消息
        String exchangeName = "E1";
        // 定义交换机模式
        channel.exchangeDeclare(exchangeName, "fanout");
        // 将对列与交换机绑定
        channel.queueBind(queueName, exchangeName, "");
        // 只允许一次执行一个消息
        channel.basicQos(1);

        // 定义消费者
        QueueingConsumer consumer = new QueueingConsumer(channel);
        // 将消费者与队列进行绑定
        /**
         * 定义回复方式 autoAck为false表示手动返回
         */
        channel.basicConsume(queueName, false, consumer);
        System.out.println("消费者1,启动。。。");
        // 获取消息
        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String msg = "消费者1收到 :" + new String(delivery.getBody());
            System.out.println(msg);
            // 告知rabbitMq当前消费的是哪一个消息
            /**
             * multiple false表示不扩展
             */
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    }

    // 消费者2
    @Test
    public void consumer2() throws Exception {
        // 创建通道
        Channel channel = connection.createChannel();
        // 定义对列
        String queueName = "counsumer2";
        channel.queueDeclare(queueName, false, false, false, null);
        // 订阅E1交换机的消息
        String exchangeName = "E1";
        // 定义交换机模式
        channel.exchangeDeclare(exchangeName, "fanout");
        // 将对列与交换机绑定
        channel.queueBind(queueName, exchangeName, "");
        // 只允许一次执行一个消息
        channel.basicQos(1);

        // 定义消费者
        QueueingConsumer consumer = new QueueingConsumer(channel);
        // 将消费者与队列进行绑定
        /**
         * 定义回复方式 autoAck为false表示手动返回
         */
        channel.basicConsume(queueName, false, consumer);
        System.out.println("消费者2,启动。。。");
        // 获取消息
        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String msg = "消费者2收到 :" + new String(delivery.getBody());
            System.out.println(msg);
            // 告知rabbitMq当前消费的是哪一个消息
            /**
             * multiple false表示不扩展
             */
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    }

}

消费者1,启动。。。

消费者2,启动。。。

消息发送成功!!!

消费者1,启动。。。
消费者1收到 :今天学校下雨了!!!——发布订阅模式,这是第0次通知
消费者1收到 :今天学校下雨了!!!——发布订阅模式,这是第1次通知
消费者1收到 :今天学校下雨了!!!——发布订阅模式,这是第2次通知
消费者1收到 :今天学校下雨了!!!——发布订阅模式,这是第3次通知
消费者1收到 :今天学校下雨了!!!——发布订阅模式,这是第4次通知
消费者1收到 :今天学校下雨了!!!——发布订阅模式,这是第5次通知

消费者2,启动。。。
消费者2收到 :今天学校下雨了!!!——发布订阅模式,这是第0次通知
消费者2收到 :今天学校下雨了!!!——发布订阅模式,这是第1次通知
消费者2收到 :今天学校下雨了!!!——发布订阅模式,这是第2次通知
消费者2收到 :今天学校下雨了!!!——发布订阅模式,这是第3次通知
消费者2收到 :今天学校下雨了!!!——发布订阅模式,这是第4次通知
消费者2收到 :今天学校下雨了!!!——发布订阅模式,这是第5次通知

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

推荐阅读更多精彩内容