简介
RabbitMQ是消息代理：它接受并转发消息。
您可以将其视为邮局：将您要发布的邮件放在邮箱中时，可以确保邮递员最终将邮件传递给您的收件人。
以此类推，RabbitMQ是一个邮政信箱，一个邮局和一个邮递员。
RabbitMQ与邮局之间的主要区别在于，它不处理纸张，而是接收，存储和转发数据消息的二进制数据。
AMQP协议
AMQP，即Advanced Message Queuing Protocol，一个提供统一消息服务的应用层标准高级消息队列协议，是应 用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户 端/中间件同产品、不同的开发语言等条件的限制。
AMQP的实现有：RabbitMQ、OpenAMQ、Apache Qpid、Redhat Enterprise MRG、AMQP Infrastructure、 ØMQ、Zyre等。
RabbitMQ和一般的消息传递使用一些术语:
1. 生产者
生产仅意味着发送。发送消息的程序是生产者：
image.png
2. 队列
队列是RabbitMQ内部的邮政信箱的名称。
许多生产者可以发送进入一个队列的消息，许多消费者可以尝试从一个队列接收数据。、
image.png
3. 消费者
消费与接收具有相似的含义。一个消费者是一个程序，主要是等待接收信息：
image.png
4. 注意事项
生产者，消费者和经纪人不必位于同一主机上。
一个应用程序既可以是生产者，也可以是消费者。
Hello World 简单示例
1. 流程图
P是我们的生产者，C是我们的消费者。中间的框是一个队列。
image.png
2. Maven依赖
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.2.0</version>
</dependency>
3.Send.java
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.nio.charset.StandardCharsets;
public class Send {
private final static String QUEUE_NAME = "hello";
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
// 建立连接
// 建立消息通道
try (Connection connection = factory.newConnection();
Channel channel = connection.createChannel()) {
//声明队列
// String queue 队列名称
// boolean durable, 是否是否持久化
// boolean exclusive,是否排他性队列。排他性队列只能在声明它的Connection中使用，连接断开时自动删除
// boolean autoDelete,是否自动删除。如果为true，至少有一个消费者连接到这个队列，
// 之后所有与这个队列连接 的消费者都断开时，队列会自动删除。
// Map<String, Object> arguments：队列的其他属性，
// 例如x-message-ttl、x-expires、x-max-length、x-maxlength-bytes、
// x-dead-letter-exchange、x-dead-letter-routing-key、x-max-priority。
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
String message = "Hello World!";
// 发送消息（发送到默认交换机AMQP Default，Direct）
// 如果有一个队列名称跟Routing Key相等，那么消息会路由到这个队列
// String exchange, String routingKey, BasicProperties props, byte[] body
channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8));
// 沉睡20秒便于观察创建的connection和channel
Thread.sleep(20000);
System.out.println(" [x] Sent '" + message + "'");
}
}
}
3.Recv.java
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
public class Recv {
private final static String QUEUE_NAME = "hello";
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
System.out.println(" [x] Received '" + message + "'");
};
//1、队列名称 2 是否自动应答,3 消费者回调函数
channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
}
}
4. 运行Send.java后进入管理页面
-
可以看到有一条连接。
image.png -
可以看到有一条管道。
image.png -
可以看到默认交换器有数据流过。
image.png -
可以看到多了一条队列，队列中有一条消息，即我们刚发的。
image.png
4. 运行Recv.java
-
控制台输出收到的消息。
image.png -
队列中的消息也被消费掉了。
image.png
5. 第二种消费方法实现消费者接口：MyConsumer.java
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownSignalException;
import java.io.IOException;
public class MyConsumer implements Consumer {
/**
* Called when consumer is registered.
*/
public void handleConsumeOk(String consumerTag) {
System.out.println("Consumer "+consumerTag +" registered");
}
/**
* Called when new message is available.
*/
public void handleDelivery(String consumerTag, Envelope env,
AMQP.BasicProperties props, byte[] body) throws IOException {
String routingKey = env.getRoutingKey(); // 队列名称
String contentType = props.getContentType(); // 内容类型
String content = new String(body, "utf-8"); // 消息正文
System.out.println("消息正文：" + content);
// 手动确认消息【参数说明：参数一：该消息的index；
// 参数二：是否批量应答，true批量确认小于index的消息】
// channel.basicAck(envelope.getDeliveryTag(), false);
}
public void handleCancel(String consumerTag) {}
public void handleCancelOk(String consumerTag) {}
public void handleRecoverOk(String consumerTag) {}
public void handleShutdownSignal(String consumerTag, ShutdownSignalException arg1) {}
}
Recv.java
import com.rabbitmq.client.*;
import java.io.IOException;
public class Recv {
private final static String QUEUE_NAME = "hello";
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
// DeliverCallback deliverCallback = (consumerTag, delivery) -> {
// String message = new String(delivery.getBody(), "UTF-8");
// System.out.println(" [x] Received '" + message + "'");
// };
// //1、队列名称 2 是否自动应答,3 消费者回调函数
// channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
//1、队列名称 2 是否自动应答,3 消费者信息
channel.basicConsume(QUEUE_NAME,true, new MyConsumer());
}
}
image.png
