一、准备工作：
创建一个spring boot项目，并且引入相关依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
二、Hello World：
Hello World
在上图的模型中，P表示消息生产者，即要发送消息的程序。C表示消费者，即处理消息的程序，消费者会一直等待消息的到来。图中红色部分为消息队列，类似于一个邮箱，可以缓存消息，生产者向其中投递消息，消费者从中取出消息。
1.生产者
public static void main(String[] args) throws IOException, TimeoutException {
//创建连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("127.0.0.1");
connectionFactory.setUsername("test");
connectionFactory.setPassword("test");
connectionFactory.setVirtualHost("test");
Connection connection = connectionFactory.newConnection();
// System.out.println(connection);
//创建通道
Channel channel = connection.createChannel();
//参数1：队列名称 参数2: 是否持久化 参数3:是否独占队列 参数4:是否⾃动删除 参数5:其他属性
channel.queueDeclare("hello", true, false, false, null);
//发送消息，此模式下不需要经过交换机。
channel.basicPublish("", "hello", null, "hello rabbitmq".getBytes());
channel.close();
connection.close();
}
运行程序，在web管理界面可以看到如下 ：
web管理界面
2.消费者
public static void main(String[] args) throws IOException, TimeoutException {
//创建连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("127.0.0.1");
connectionFactory.setUsername("test");
connectionFactory.setPassword("test");
connectionFactory.setVirtualHost("test");
Connection connection = connectionFactory.newConnection();
// System.out.println(connection);
//创建通道
Channel channel = connection.createChannel();
//绑定消息队列
channel.queueDeclare("hello", true, false, false, null);
//获取消息并处理
channel.basicConsume("hello", true, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println(new String(body));
}
});
}
程序运行结果
web管理界面
三、Work queues：
任务模型
消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。此时就可以使用work模型：让多个消费者绑定到⼀个队列，共同消费队列中的消息。队列中的消息一旦消费，就会消失，因此任务是不会被重复执行的。
1.生产者
//发送消息，此模式下不需要经过交换机。
for (int i = 0; i < 10; i++) {
channel.basicPublish("", "hello", null, ("第" + i + "条消息").getBytes());
}
2.消费者
//编写两个消费者
//获取消息并处理
channel.basicConsume("hello", true, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者2:" + new String(body));
}
});
注意：先运行消费者，再运行生产者。如果先运行生产者，在运行第一个消费者的时候，有可能已经把消息处理完。
结果
通过结果可以看出，消费者1获得了编号为偶数的消息，消费者2获得了编号为奇数的消息。默认情况下，RabbitMQ将按顺序将每个消息发送给下⼀个使用者。平均而⾔，每个消费者都会收到相同数量的消息。这种分发消息的方式称为循环。
3.消息自动确认机制
消费者1：
//设置⼀次只接受⼀条未确认的消息
channel.basicQos(1);
//获取消息并处理 参数2:是否⾃动确认消息
channel.basicConsume("hello", false, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
try {
Thread.sleep(500); //每次都让消费者1睡眠一会
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("消费者1:" + new String(body));
channel.basicAck(envelope.getDeliveryTag(), false);//⼿动确认消息
}
});
消费者2：
//设置⼀次只接受⼀条未确认的消息
channel.basicQos(1);
//获取消息并处理 参数2:是否⾃动确认消息
channel.basicConsume("hello", false, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者2:" + new String(body));
channel.basicAck(envelope.getDeliveryTag(), false);//⼿动确认消息
}
});
运行结果
从运行结果可以看出，消息不再平均分配给两个消费者，而是根据处理的能力去分配。
四、Publish/Subscribe（广播模型）：
广播模型
在广播模式下，消息发送流程是这样的：
- 可以有多个消费者。
- 每个消费者有自己的队列。
- 每个队列都要绑定到交换机。
- 生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。
- 交换机把消息发送给绑定过的所有队列。
- 队列的消费者都能拿到消息，实现⼀条消息被多个消费者消费。
1.生产者
//从连接工具获取连接。
Channel channel = ConnectionUtil.GetConnection().createChannel();
//声明交换机
channel.exchangeDeclare("logs", "fanout");//广播⼀条消息多个消费者同时消费
//发布消息
channel.basicPublish("logs", "", null, "hello".getBytes());
2.消费者
Channel channel = ConnectionUtil.GetConnection().createChannel();
//绑定交换机
channel.exchangeDeclare("logs", "fanout");
//创建临时队列
String queue = channel.queueDeclare().getQueue();
//将临时队列绑定exchange
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("消费者1: " + new String(body));
}
});
编写三个消费者，同时运行三个消息者和生产者
三个消费者同时获得消息
