waitzkj 2020-06-06
fanout 扇出 也曾为广播
在广播模式下,消息发送流程是这样的:
生产者
public class Provider { public static void main(String[] args) throws IOException { Connection connection = rabbitMQUtils.getConnection(); Channel channel = connection.createChannel(); //将通道声明指定交换机 //参数1交换机名字 参数2交换机类型 fanout 广播类型 channel.exchangeDeclare("logs","fanout"); //发送消息 channel.basicPublish("logs","",null,"fanout type message".getBytes()); rabbitMQUtils.connectionAndchannelClose(connection,channel); } }
消费者(创建多个相同的消费者)
public class Customer1 { public static void main(String[] args) throws IOException { Connection connection = rabbitMQUtils.getConnection(); Channel channel = connection.createChannel(); //通道绑定交换机 channel.exchangeDeclare("logs","fanout"); //临时列队 String queue = channel.queueDeclare().getQueue(); //绑定交换机和队列 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 { super.handleDelivery(consumerTag, envelope, properties, body); System.out.println("消费者-1"+new String(body)); } }); } }