下图是RabbitMQ的基本结构:
组成部分说明如下:
消息发布接收流程:
-----发送消息-----
----接收消息-----
RabbitMQ由Erlang语言开发,Erlang语言用于并发及分布式系统的开发,在电信领域应用广泛,OTP(Open Telecom Platform)作为Erlang语言的一部分,包含了很多基于Erlang开发的中间件及工具库,安装RabbitMQ需 要安装Erlang/OTP,并保持版本匹配,如下图:
RabbitMQ的下载地址:http://www.rabbitmq.com/download.html
1)下载erlang 地址如下: http://erlang.org/download/otp_win64_20.3.exe
erlang安装完成需要配置erlang环境变量: ERLANG_HOME=D:\Program Files\erl9.3
在path中添加%ERLANG_HOME%\bin;
2)安装RabbitMQ https://github.com/rabbitmq/rabbitmq-server/releases/tag/v3.7.3
安装成功后会自动创建RabbitMQ服务并且启动。
1)从开始菜单启动RabbitMQ 完成在开始菜单找到RabbitMQ的菜单:
RabbitMQ Service-install
:安装服务RabbitMQ Service-remove
删除服务RabbitMQ Service-start
启动RabbitMQ Service-stop
停止2)如果没有开始菜单则进入安装目录下sbin目录手动启动:
1)安装并运行服务
rabbitmq-service.bat install
安装服务rabbitmq-service.bat stop
停止服务rabbitmq-service.bat start
启动服务2)安装管理插件
安装rabbitMQ的管理插件,方便在浏览器端管理RabbitMQ
管理员身份运行 rabbitmq-plugins.bat enable rabbitmq_management
3、启动成功 登录RabbitMQ 进入浏览器,输入:http://localhost:15672
初始账号和密码:guest/guest
按照官方教程(http://www.rabbitmq.com/getstarted.html)测试hello world:
1)java client 生产者和消费者都属于客户端,rabbitMQ的java客户端如下:
我们先用 rabbitMQ官方提供的java client测试,目的是对RabbitMQ的交互过程有个清晰的认识。 参考 :https://github.com/rabbitmq/rabbitmq-java-client/
2)创建maven工程
创建生产者工程和消费者工程,分别加入RabbitMQ java client的依赖。
test-rabbitmq-producer
:生产者工程
test-rabbitmq-consumer
:消费者工程
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp‐client</artifactId>
<version>4.0.3</version>
<!‐‐此版本与spring boot 1.5.9版本匹配‐‐>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring‐boot‐starter‐logging</artifactId>
</dependency>
public class Producer01 {
//队列名称
private static final String QUEUE = "helloworld";
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = null;
Channel channel = null;
try {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
factory.setUsername("guest");
factory.setPassword("guest");
factory.setVirtualHost("/");
//rabbitmq默认虚拟机名称为“/”,虚拟机相当于一个独立的mq服务 器
// 创建与RabbitMQ服务的TCP连接
connection = factory.newConnection();
//创建与Exchange的通道,每个连接可以创建多个通道,每个通道代表一个会话任务
channel = connection.createChannel(); /*** 声明队列,如果Rabbit中没有此队列将自动创建 * param1:队列名称 * param2:是否持久化 * param3:队列是否独占此连接 * param4:队列不再使用时是否自动删除此队列 * param5:队列参数 */
channel.queueDeclare(QUEUE, true, false, false, null);
String message = "helloworld小明" + System.currentTimeMillis();
/*** 消息发布方法 * param1:Exchange的名称,如果没有指定,则使用Default Exchange * param2:routingKey,消息的路由Key,是用于Exchange(交换机)将消息转发到指定的消息队列 * param3:消息包含的属性 * param4:消息体 */
/*** 这里没有指定交换机,消息将发送给默认交换机,每个队列也会绑定那个默认的交换机,但是不能显 示绑定或解除绑定* 默认的交换机,routingKey等于队列名称 */
channel.basicPublish("", QUEUE, null, message.getBytes());
System.out.println("Send Message is:'" + message + "'");
} catch (Exception ex) {
ex.printStackTrace();
} finally {
if (channel != null) {
channel.close();
}
if (connection != null) {
connection.close();
}
}
}
}
public class Consumer01 {
private static final String QUEUE = "helloworld";
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
//设置MabbitMQ所在服务器的ip和端口
factory.setHost("127.0.0.1");
factory.setPort(5672);
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
//声明队列
channel.queueDeclare(QUEUE, true, false, false, null);
//定义消费方法
DefaultConsumer consumer = new DefaultConsumer(channel) {
/*** 消费者接收消息调用此方法
* @param consumerTag 消费者的标签,在channel.basicConsume()去指定
* @param envelope 消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志 (收到消息失败后是否需要重新发送)
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//交换机
String exchange = envelope.getExchange();
//路由key
String routingKey = envelope.getRoutingKey();
//消息id
long deliveryTag = envelope.getDeliveryTag();
//消息内容
String msg = new String(body, "utf‐8");
System.out.println("receive message.." + msg);
}
};/*** 监听队列String queue, boolean autoAck,Consumer callback * 参数明细 * 1、队列名称 * 2、是否自动回复,设置为true为表示消息接收到自动向mq回复接收到了,mq接收到回复会删除消息,设置 为false则需要手动回复 * 3、消费消息的方法,消费者接收到消息后调用此方法 */
channel.basicConsume(QUEUE, true, consumer);
}
}
1、发送端操作流程 1)创建连接 2)创建通道 3)声明队列 4)发送消息
2、接收端 1)创建连接 2)创建通道 3)声明队列 4)监听队列 5)接收消息 6)ack回复