|
|
|
package com.sunyo.wlpt.message.bus.service.rabbit.utils;
|
|
|
|
|
|
|
|
import com.rabbitmq.client.*;
|
|
|
|
import com.sunyo.wlpt.message.bus.service.utils.IdUtils;
|
|
|
|
import lombok.extern.slf4j.Slf4j;
|
|
|
|
import org.springframework.beans.factory.annotation.Autowired;
|
|
|
|
import org.springframework.beans.factory.annotation.Value;
|
|
|
|
import org.springframework.data.redis.core.StringRedisTemplate;
|
|
|
|
import org.springframework.stereotype.Component;
|
|
|
|
|
|
|
|
import javax.annotation.PostConstruct;
|
|
|
|
import java.io.IOException;
|
|
|
|
import java.util.concurrent.TimeUnit;
|
|
|
|
import java.util.concurrent.TimeoutException;
|
|
|
|
|
|
|
|
/**
|
|
|
|
* @author 子诚
|
|
|
|
* Description：
|
|
|
|
* 时间：2020/7/21 9:32
|
|
|
|
*/
|
|
|
|
@Slf4j
|
|
|
|
@Component
|
|
|
|
public class DirectUtils {
|
|
|
|
|
|
|
|
@Autowired
|
|
|
|
private StringRedisTemplate redisTemplate;
|
|
|
|
|
|
|
|
@Value("${spring.rabbitmq.host}")
|
|
|
|
private String host;
|
|
|
|
|
|
|
|
@Value("${spring.rabbitmq.port}")
|
|
|
|
private int port;
|
|
|
|
|
|
|
|
@Value("${spring.rabbitmq.username}")
|
|
|
|
private String username;
|
|
|
|
|
|
|
|
@Value("${spring.rabbitmq.password}")
|
|
|
|
private String password;
|
|
|
|
|
|
|
|
@Value("${spring.rabbitmq.virtual-host}")
|
|
|
|
private String vHost;
|
|
|
|
|
|
|
|
/**
|
|
|
|
* @return 链接 rabbitmq
|
|
|
|
* @throws IOException IO异常
|
|
|
|
* @throws TimeoutException 超时异常
|
|
|
|
*/
|
|
|
|
@PostConstruct
|
|
|
|
public Connection getConnection() throws IOException, TimeoutException
|
|
|
|
{
|
|
|
|
//定义连接工厂
|
|
|
|
ConnectionFactory factory = new ConnectionFactory();
|
|
|
|
//设置服务地址
|
|
|
|
factory.setHost(host);
|
|
|
|
//端口,amqp协议 端口 类似与mysql的3306
|
|
|
|
factory.setPort(port);
|
|
|
|
//设置账号信息，用户名、密码、vhost
|
|
|
|
factory.setVirtualHost(vHost);
|
|
|
|
factory.setUsername(username);
|
|
|
|
factory.setPassword(password);
|
|
|
|
// 通过工程获取连接
|
|
|
|
Connection connection = factory.newConnection();
|
|
|
|
return connection;
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* 链接 RabbitMQ
|
|
|
|
*
|
|
|
|
* @param hostIp mq服务器Ip地址
|
|
|
|
* @param hostPort mq服务器端口号
|
|
|
|
* @param vHostName VirtualHost名称
|
|
|
|
* @param userName 登录账号
|
|
|
|
* @param password 登录密码
|
|
|
|
* @return 返回链接
|
|
|
|
* @throws Exception
|
|
|
|
*/
|
|
|
|
public static Connection getConnection(String hostIp, int hostPort, String vHostName, String userName, String password) throws Exception
|
|
|
|
{
|
|
|
|
//定义连接工厂
|
|
|
|
ConnectionFactory factory = new ConnectionFactory();
|
|
|
|
//设置服务地址
|
|
|
|
factory.setHost(hostIp);
|
|
|
|
//端口
|
|
|
|
factory.setPort(hostPort);
|
|
|
|
//设置账号信息，用户名、密码、vhost
|
|
|
|
factory.setVirtualHost(vHostName);
|
|
|
|
factory.setUsername(userName);
|
|
|
|
factory.setPassword(password);
|
|
|
|
// 通过工程获取连接
|
|
|
|
Connection connection = factory.newConnection();
|
|
|
|
return connection;
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* 关闭通道和关闭连接的工具方法
|
|
|
|
*
|
|
|
|
* @param channel 通道
|
|
|
|
* @param conn 连接
|
|
|
|
*/
|
|
|
|
public static void closeConnectionAndChanel(Channel channel, Connection conn)
|
|
|
|
{
|
|
|
|
try {
|
|
|
|
if (channel != null) {
|
|
|
|
channel.close();
|
|
|
|
}
|
|
|
|
if (conn != null) {
|
|
|
|
conn.close();
|
|
|
|
}
|
|
|
|
} catch (Exception e) {
|
|
|
|
e.printStackTrace();
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
* DirectExchange的 消息生产者
|
|
|
|
*
|
|
|
|
* @param exchangeName 交换机名称
|
|
|
|
* @param routingKeyName 路由键名称
|
|
|
|
* @param msg 发送的消息
|
|
|
|
* @throws IOException
|
|
|
|
* @throws TimeoutException
|
|
|
|
*/
|
|
|
|
public void directProducer(String exchangeName, String routingKeyName, String msg) throws IOException, TimeoutException
|
|
|
|
{
|
|
|
|
// 1、创建ConnectionFactory
|
|
|
|
Connection connection = getConnection();
|
|
|
|
// 2、 通过Connection创建一个新的Channel
|
|
|
|
Channel channel = connection.createChannel();
|
|
|
|
// 3、开启消息的确认机制（confirm:保证消息能够发送到 exchange）
|
|
|
|
channel.confirmSelect();
|
|
|
|
// 4、避免消息被重复消费
|
|
|
|
AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
|
|
|
|
// 指定消息是否需要持久化，1：需要持久化；2：不需要持久化
|
|
|
|
.deliveryMode(1)
|
|
|
|
// 设置全局唯一消息机制id(雪花id)
|
|
|
|
.messageId(IdUtils.generateId())
|
|
|
|
.build();
|
|
|
|
// 5、开启 return 机制(保证消息，从 Exchange 分发到 Queue )
|
|
|
|
channel.addReturnListener(new ReturnListener() {
|
|
|
|
@Override
|
|
|
|
public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException
|
|
|
|
{
|
|
|
|
// 当消息没有从 Exchange 分发到 Queue 时，才会执行
|
|
|
|
log.error(new String(body, "UTF8") + "->没有从 Exchange 分发到Queue中");
|
|
|
|
}
|
|
|
|
});
|
|
|
|
|
|
|
|
// 6、发送消息,并指定 mandatory 参数为true
|
|
|
|
channel.basicPublish(exchangeName, routingKeyName, true, properties, msg.getBytes());
|
|
|
|
log.info("消息生产者，目标交换机：{}；路由键：{}；发送信息：{}", exchangeName, routingKeyName, msg);
|
|
|
|
// 7、添加一个异步 confirm 确认监听，用于发送消息到Broker端之后，回送消息的监听
|
|
|
|
channel.addConfirmListener(new ConfirmListener() {
|
|
|
|
// 发送成功
|
|
|
|
@Override
|
|
|
|
public void handleAck(long deliveryTag, boolean multiple) throws IOException
|
|
|
|
{
|
|
|
|
log.info("消息发送成功，标识：{}；是否是批量：{}", deliveryTag, multiple);
|
|
|
|
}
|
|
|
|
|
|
|
|
// 发送失败
|
|
|
|
@Override
|
|
|
|
public void handleNack(long deliveryTag, boolean multiple) throws IOException
|
|
|
|
{
|
|
|
|
log.error("消息发送失败，标识：{}；是否是批量：{}", deliveryTag, multiple);
|
|
|
|
}
|
|
|
|
});
|
|
|
|
// finally，关闭连接
|
|
|
|
closeConnectionAndChanel(channel, connection);
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* DirectExchange的 消息消费者
|
|
|
|
*
|
|
|
|
* @throws IOException IO异常
|
|
|
|
* @throws TimeoutException 超时异常
|
|
|
|
*/
|
|
|
|
public void directConsumer(String queueName, String exchangeName, String routingKeyName) throws IOException, TimeoutException
|
|
|
|
{
|
|
|
|
// 1、创建ConnectionFactory
|
|
|
|
Connection connection = getConnection();
|
|
|
|
// 2、 通过Connection创建一个新的Channel
|
|
|
|
Channel channel = connection.createChannel();
|
|
|
|
// 3、设置绑定关系（队列、交换机名称、路由键名称）
|
|
|
|
channel.queueBind(queueName, exchangeName, routingKeyName);
|
|
|
|
// 一次只接受一条未确认的消息
|
|
|
|
channel.basicQos(1);
|
|
|
|
// 4、开启监听Queue
|
|
|
|
DefaultConsumer consumer = new DefaultConsumer(channel) {
|
|
|
|
@Override
|
|
|
|
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException
|
|
|
|
{
|
|
|
|
try {
|
|
|
|
// 0、获取出全局唯一的 信息业务id(messageId)
|
|
|
|
String messageId = properties.getMessageId();
|
|
|
|
// 必须保证 messageId 不为空，避免空指针异常
|
|
|
|
if (redisTemplate.opsForValue().setIfAbsent(messageId, "0", 10, TimeUnit.MINUTES)) {
|
|
|
|
|
|
|
|
// 消费成功，将redis中的 messageId 对应的value修改为 1
|
|
|
|
redisTemplate.opsForValue().set(messageId, "1", 10, TimeUnit.MINUTES);
|
|
|
|
// 手动ack
|
|
|
|
channel.basicAck(envelope.getDeliveryTag(), false);
|
|
|
|
|
|
|
|
log.info("接收到消息：" + new String(body, "UTF-8"));
|
|
|
|
} else {
|
|
|
|
// 获取redis中的value，如果是1，就手动ack。如果是0，就什么也不做（是0代表着，正在被消费中）
|
|
|
|
if ("1".equalsIgnoreCase(redisTemplate.opsForValue().get(messageId))) {
|
|
|
|
// 手动ack
|
|
|
|
channel.basicAck(envelope.getDeliveryTag(), false);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
} catch (Exception e) {
|
|
|
|
// 手动ack
|
|
|
|
channel.basicAck(envelope.getDeliveryTag(), false);
|
|
|
|
log.error("接收消息发送错误：" + e.getMessage());
|
|
|
|
}
|
|
|
|
}
|
|
|
|
};
|
|
|
|
// 消费消息
|
|
|
|
channel.basicConsume(queueName, false, consumer);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
...
|
...
|
|
