其他分享
首页 > 其他分享> > RabbitMQ学习(三)广播模型(fanout)

RabbitMQ学习(三)广播模型(fanout)

作者:互联网

一、消息发送流程
在这里插入图片描述
广播模式下,消息的发送流程:

二、代码实现
生产者代码:

package fanout;

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

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @author wmh
 * @date 2021/1/11
 */
public class Provider {
    public static void main(String[] args) throws IOException, TimeoutException {
//        创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("主机名");
        connectionFactory.setPort(5672);
//        绑定虚拟主机
        connectionFactory.setVirtualHost("/ems");
        connectionFactory.setUsername("ems");
        connectionFactory.setPassword("123");
//        创建连接
        Connection connection = connectionFactory.newConnection();
//        创建连接通道
        Channel channel = connection.createChannel();
//        绑定交换机
//        参数1:交换机的名字 参数2:交换机的类型
        channel.exchangeDeclare("logs","fanout");
//        发送消息
//        参数1:交换机的名字 参数2:routingKey 参数3:队列消息的持久化设置 参数4:要发送的消息
        channel.basicPublish("logs","",null,"fanout messages".getBytes());
//        关闭连接
        channel.close();
        connection.close();
    }
}

消费者代码(创建了三个消费者,只展示其中一个,其他代码消费者相同):

package fanout;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @author wmh
 * @date 2021/1/11
 */
public class Customer1 {
    public static void main(String[] args) throws IOException, TimeoutException {
        //        创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("主机名");
        connectionFactory.setPort(5672);
//        绑定虚拟主机
        connectionFactory.setVirtualHost("/ems");
        connectionFactory.setUsername("ems");
        connectionFactory.setPassword("123");
//        创建连接
        Connection connection = connectionFactory.newConnection();
//        创建连接通道
        Channel channel = connection.createChannel();
//        将通道绑定给交换机
        channel.exchangeDeclare("logs","fanout");
//        channel.exchangeBind
//        创建临时队列
        String queue = channel.queueDeclare().getQueue();
//        将通道绑定给队列(通过通道将队列与交换机进行绑定)
//        参数1:队列名称 参数2:交换机名称 参数3:routingKey
        channel.queueBind(queue,"logs","");
//        消费消息
        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("Customer1消费了消息:"+new String(body));
            }
        });
    }
}

运行结果:
在这里插入图片描述
在这里插入图片描述

标签:connectionFactory,fanout,队列,绑定,RabbitMQ,广播,交换机,import,channel
来源: https://blog.csdn.net/weixin_42765792/article/details/112492627