Java中简单使用RabbitMQ进行消息收发_java rmq发送消息

在项目中需要使用RabbitMQ作为消息队列，于是写了一个RabbitMQ的服务提供类，这个服务提供类包含RabbitMQ相关实例的初始化及建立连接、消息的接收以及消息的发送，想偷懒的童鞋拷过去改改配置什么的就可直接使用，以此做个记录，方便以后温故而知新。

Windows中安装RabbitMQ-Server

安装Erlang

因为RabbitMQ是使用Erlang开发的，所以要使用RabbitMQ-Server首先就需要安装Erlang的运行环境，可以在官网下载安装，安装后需要配置环境变量。

配置示例：

ERLANG_HOME：D:\software\erl10.1

Path末尾拼接：%ERLANG_HOME%\bin;

安装RabbitMQ-Server

可在官网自行下载安装，安装后需要配置环境变量。

配置示例：

RABBITMQ_SERVER：D:\software\RabbitMQ Server\rabbitmq_server-3.7.9

Path末尾拼接：%RABBITMQ_SERVER%\sbin;

配置完成后便可启动RabbitMQ-Server，初始用户名与密码是guest：guest。

其他具体的配置及命令，不详述，请自行查找。

需要使用的依赖包

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
</dependency>
1
2
3
4

RabbitMQ服务提供类，含消息接收与发送

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

import javax.annotation.PostConstruct;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.MessageProperties;

/**
 * rabbitmq服务提供类
 * @author dr
 */
@Component
public class RabbitMQProvider {
	private Logger logger = LoggerFactory.getLogger(this.getClass());
	/**是否启用RabbitMQ*/
	@Value("${rabbitmq.setting.enable}")
	private boolean enable;
	/**交换器名*/
	@Value("${rabbitmq.setting.exchange.name}")
	private String exchangeName;
	/**下行队列名*/
	@Value("down.${push.setting.school}")
	private String downQueueName;
	@Value("down.${push.setting.school}")
	private String downRoutingKey;
	@Value("up.${push.setting.school}")
	private String upQueueName;
	@Value("up.${push.setting.school}")
	private String upRoutingKey;
	@Value("${rabbitmq.setting.username}")
	private String username;
	@Value("${rabbitmq.setting.password}")
	private String password;
	@Value("${rabbitmq.setting.host}")
	private String host;
	@Value("${rabbitmq.setting.port}")
	private Integer port;
    // 是否自动应答
	private boolean autoAck = true;
	private Channel channelDown;
	private Channel channelUp;
	private Connection upConnection;
	private Connection downConnection;
	@Value("${rabbitmq.setting.message-life-time}")
	private Integer messageLifeTime;
	/**处理接收到的消息的处理实例，即我们的业务代码*/
	@Autowired
	private MQDealer mqDealer;
	
    // 初始化
	@PostConstruct
	private void init() {
		if (!enable) return;
		
		try {
			// 创建连接
			ConnectionFactory factory = new ConnectionFactory();
			factory.setUsername(username);
			factory.setPassword(password);
			factory.setHost(host);
			factory.setPort(port);
			// 队列参数
			Map<String, Object> args = new HashMap<>();
			args.put("x-message-ttl", messageLifeTime*1000);// 消息过期时间
			
			// 创建上行连接
			upConnection = factory.newConnection();
			// 创建上行通道
			channelUp = upConnection.createChannel();
			// 声明创建配置上行队列
			channelUp.queueDeclare(upQueueName, true, false, false, args);
			// 将队列与交换器绑定，并设置路由码
			channelUp.queueBind(upQueueName, exchangeName, upRoutingKey);
			
			downConnection = factory.newConnection();
			channelDown = downConnection.createChannel();
			channelDown.queueDeclare(downQueueName, true, false, false, args);
			channelDown.queueBind(downQueueName, exchangeName, downRoutingKey);
			
			receiveMessage();
		} catch (Exception e) {
			logger.error("启动MQ下行通道时出现异常！", e);
		}
	}

	/**
	 * 持续监听队列以接收数据
	 * @throws IOException
	 * @throws TimeoutException
	 */
	private void receiveMessage() throws IOException, TimeoutException {
		// 每次缓存5个消息在本地
		channelDown.basicQos(5);
		channelDown.basicConsume(downQueueName, autoAck, "myConsumerTag", new DefaultConsumer(channelDown) {
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
					byte[] body) throws IOException {
				String message = new String(body, "UTF-8");
				logger.debug(
						downQueueName + " Received '" + message + "'" + ", routingKey: " + envelope.getRoutingKey());
				 // 处理接收到的消息
				 mqDealer.deal(message);
				// 持续监听
				channelDown.basicConsume(downQueueName, autoAck, "myConsumerTag", this);
				channelDown.basicAck(envelope.getDeliveryTag(), true);
			}
		});
	}

	/**
	 * 向上行消息队列发送一条消息
	 * @param message
	 * @throws IOException
	 * @throws TimeoutException
	 */
	public void sendMessage(String message) throws IOException, TimeoutException {
		channelUp.basicPublish(exchangeName, upRoutingKey, true, MessageProperties.TEXT_PLAIN, message.getBytes());
		
		logger.debug("send message to " + upQueueName + ": " + message);
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136

解析：

• 因为网友反映使用单连接单通道同时发送和接收消息容易导致队列阻塞，所以这里采用了双连接双通道，分别负责接收和发送；
• 这里使用rabbitMQ的方式是直接使用的rabbitMQ提供的开发包，而非Spring整合过的；
• 建议设置消息过期时间，因为太多消息积压在队列中而没有消费的话，可能会导致队列的阻塞，我在开发中已经碰到了一次；
• 在声明队列时，如果该队列不存在则会创建，如果已经存在则直接使用现有的，需要注意的是，如果已经有现存队列，那么再次声明时，需要属性和现存的完全一致，否则会出现异常；
• 这段代码中属性的加载，连接、通道等实例的初始化使用了Spring的IoC，如果项目没有使用Spring，需要自行加载和初始执行；