package com.ks.push.manager.rabbitmq;
|
|
import com.google.gson.Gson;
|
import com.rabbitmq.client.Channel;
|
import org.slf4j.Logger;
|
import org.slf4j.LoggerFactory;
|
import org.springframework.amqp.core.Message;
|
import org.springframework.amqp.rabbit.annotation.RabbitListener;
|
import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
import org.springframework.stereotype.Component;
|
import org.yeshi.utils.StringUtil;
|
|
import javax.annotation.Resource;
|
import java.nio.charset.StandardCharsets;
|
|
/**
|
* @author hxh
|
* @title: RabbitDelayConsumer
|
* @description: 延时队列消费处理
|
* @date 2024/10/12 11:21
|
*/
|
@Component
|
public class RabbitDelayConsumer {
|
|
private Logger logger = LoggerFactory.getLogger("infoLog");
|
|
@Resource
|
private RabbitTemplate rabbitTemplate;
|
|
|
@RabbitListener(queues = RabbitmqSenderUtil.DELAY_QUEUE_NAME, ackMode = "MANUAL")
|
public void onMessage(Message message, Channel channel) throws Exception {
|
String msg = new String( message.getBody(), StandardCharsets.UTF_8);
|
logger.info("RabbitDelayConsumer-{}",msg);
|
DelayMsgInfo msgInfo = new Gson().fromJson(msg, DelayMsgInfo.class);
|
if(!StringUtil.isNullOrEmpty(msgInfo.getQueueName())){
|
// 队列消息
|
RabbitmqSenderUtil.sendQueueMsg(rabbitTemplate, msgInfo.getQueueName(), msgInfo.getMsg());
|
}else{
|
// 交换机消息
|
RabbitmqSenderUtil.sendExchangeMsg(rabbitTemplate, msgInfo.getExchangeName(), msgInfo.getRoutingKey(), msgInfo.getMsg());
|
}
|
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
|
}
|
}
|
