您的位置: 首页  >  文章  >  【RabbitMQ】1.消息队列的HelloWorld

【RabbitMQ】1.消息队列的HelloWorld

分类: 文章 • 2023-03-19 14:58:18

1.基本原理与使用

1.1.原理

【RabbitMQ】1.消息队列的HelloWorld–各组件解释：
Broker：消息队列服务进程，此进程包括两个部分：Exchange和Queue。
Exchange：消息队列交换机，按一定的规则将消息路由转发到某个队列，对消息进行过虑。
北京市昌平区建材城西路金燕龙办公楼一层 电话：400-618-9090
传智播客Queue：消息队列，存储消息的队列，消息到达队列并转发给指定的消费方。
Producer：消息生产者，即生产方客户端，生产方客户端将消息发送到MQ。
Consumer：消息消费者，即消费方客户端，接收MQ转发的消息。
消息发布接收流程：
-----发送消息-----
1、生产者和Broker建立TCP连接。
2、生产者和Broker建立通道。
3、生产者通过通道消息发送给Broker，由Exchange将消息进行转发。
4、Exchange将消息转发到指定的Queue（队列）
----接收消息-----
1、消费者和Broker建立TCP连接
2、消费者和Broker建立通道
3、消费者监听指定的Queue（队列）
4、当有消息到达Queue时Broker默认将消息推送给消费者。
5、消费者接收到消息。

1.2.下载mq与依赖

下载地址：
需要erlang：http://erlang.org/download/otp_win64_20.3.exe
mq： https://github.com/rabbitmq/rabbitmq-server/releases/tag/v3.7.15
配置erlang环境变量
ERLANG_HOME=D:\Program Files\erl9.3
在path中添加 %ERLANG_HOME%\bin

1.3.安装完成后自动创建RabbitMQ服务并启动

1）在开始菜单找
【RabbitMQ】1.消息队列的HelloWorld2）如果没有开始菜单则进入安装目录下sbin目录手动启动
【RabbitMQ】1.消息队列的HelloWorld
rabbitmq-service.bat install 安装服务
rabbitmq-service.bat stop 停止服务
rabbitmq-service.bat start 启动服务

3）安装管理插件，方便在浏览器端管理mq
以管理员身份运行 rabbitmq-plugins.bat enable rabbitmq_management

  1. 浏览器访问 http://localhost:15762
    初始 -u/-p : guest/guest

官网教程： http://www.rabbitmq.com/getstarted.html

1.4.“Hello World”

1.4.1 引入 RabbitMQ java client 依赖
maven：

<!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.7.0</version>
</dependency>

或 gradle：

// https://mvnrepository.com/artifact/com.rabbitmq/amqp-client
compile group: 'com.rabbitmq', name: 'amqp-client', version: '5.7.0'

1.4.2 创建生产者和消费者工程
test-rabbitmq-producer：生产者工程
test-rabbitmq-consumer：消费者工程

定义生产者：

public class Producer01 {
//队列名称
	private static final String QUEUE_NAME = "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");
			//rabbitmq默认虚拟机名称为“/”，每个虚拟机相当于一个独立的mq服务器
			factory.setVirtualHost("/");
			
			// 创建与RabbitMQ服务的TCP连接
			connection = factory.newConnection();
			//创建与Exchange的通道，每个连接可以创建多个通道，每个通道代表一个会话任务
			channel = connection.createChannel();
			/**
			* 声明队列，如果Rabbit中没有此队列将自动创建
			* queueDeclare参数详情：
			* String queue  队列名称
			* boolean durable: 是否持久化
			* boolean exclusive: 队列是否独占此连接
			* boolean autoDelete: 队列不再使用时是否自动删除此队列
			* arguments: 队列参数，可以设置一个队列的扩展参数，如，存活时间
			*/
			channel.queueDeclare(QUEUE_NAME , true, false, false, null);
			/**
			* 消息发布方法
			* basicPublish()参数详解
			* String exchange：交换机名称，如果没有指定(设为“”)，则使用Default Exchange
			* String routingKey:消息的路由Key,交换机根据key将消息转发到指定的队列；若使用默认交换机，key==队列名
			* BasicProperties props:消息包含的属性
			* byte[] body：消息体
			*/
			String message = "helloMsg"+System.currentTimeMillis();
			channel.basicPublish("", QUEUE_NAME, 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_NAME = "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_NAME , true, false, false, null);
		//定义消费方法
		DefaultConsumer consumer = new DefaultConsumer(channel) {
			/**
			* 消费者接收消息调用此方法
			* @param consumerTag 消费者的标签，在channel.basicConsume()去指定
			* @param envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
			* @param properties 消息属性
			* @param body 消息内容--字节数组
			* @throws IOException
			*/
			@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，mq在channel中标识消息的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、queue：队列名称
		* 2、autoAck：是否自动回复，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动回复
		* 3、callback：消费消息的方法，消费者接收到消息后调用此方法
		*/
			channel.basicConsume(QUEUE_NAME , true, consumer);
	}
}

1.4.3 总结
1、发送端流程
1）创建连接
2）创建通道
3）声明队列
4）发送消息

2、接收端流程
1）创建连接
2）创建通道
3）声明队列
4）监听队列
5）接收消息
6）ack回复

相关推荐