先粘代码
生产者
ConnectionFactory connectionFactory = new ConnectionFactory();
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME,BuiltinExchangeType.FANOUT);
channel.basicPublish(EXCHANGE_NAME,"",null,"你是不是傻".getBytes());
channel.close();
connection.close();
消费者
class Consume{
static Channel channel;
private final static String QUEUE_NAME = "test_queue_fanout_1";
private final static String EXCHANGE_NAME = "fanout";
public static void main(String[] args) {
try {
ConnectionFactory connectionFactory = new ConnectionFactory();
Connection connection = connectionFactory.newConnection();
channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
String queueName = channel.queueDeclare().getQueue();
System.out.println(queueName);
System.out.println(QUEUE_NAME);
channel.queueBind(queueName,EXCHANGE_NAME,"");
Consumer consumer = 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(new String(body,"utf-8"));
}
};
channel.basicConsume(queueName,true,consumer);
// 监听队列,手动返回完成
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
}
}
image.png
注意channel.queueBind(queueName,EXCHANGE_NAME,""); 这里的queueName 不是之前定义的那个
上图显示 多开了几个消费者 结果每一个消费者的queueName都不一样
网友评论