武
6 years ago
7 changed files with 202 additions and 132 deletions
@ -1,4 +1,4 @@ |
|||
spring: |
|||
profiles: |
|||
active: test |
|||
include: common, util-test |
|||
active: dev |
|||
include: common, util-dev |
|||
@ -1,29 +1,33 @@ |
|||
//package com.ccsens.util.mq;
|
|||
//
|
|||
//import cn.hutool.core.date.DateUtil;
|
|||
//import com.ccsens.util.config.RabbitMQConfig;
|
|||
//import com.rabbitmq.client.AMQP;
|
|||
//import org.springframework.amqp.rabbit.annotation.RabbitHandler;
|
|||
//import org.springframework.amqp.rabbit.annotation.RabbitListener;
|
|||
//import org.springframework.amqp.support.AmqpHeaders;
|
|||
//import org.springframework.messaging.handler.annotation.Headers;
|
|||
//import org.springframework.messaging.handler.annotation.Payload;
|
|||
//import org.springframework.stereotype.Component;
|
|||
//
|
|||
//import java.util.Date;
|
|||
//import java.util.Map;
|
|||
//
|
|||
///**
|
|||
// * @description:
|
|||
// * @author: wuHuiJuan
|
|||
// * @create: 2019/12/27 15:40
|
|||
// */
|
|||
//@Component
|
|||
//public class DelayConsumer {
|
|||
// @RabbitListener(queues = RabbitMQConfig.GAME_STATUS_DELAY_SEND_QUEUE)
|
|||
// @RabbitHandler
|
|||
// public void receive(@Payload String msg, @Headers Map<String, Object> headers, AMQP.Channel channel ) {
|
|||
// System.out.println("接收到的消息:"+msg +"||" + new Date());
|
|||
// }
|
|||
//
|
|||
//}
|
|||
package com.ccsens.util.mq; |
|||
|
|||
import com.ccsens.util.config.RabbitMQConfig; |
|||
import com.rabbitmq.client.Channel; |
|||
import org.springframework.amqp.rabbit.annotation.RabbitHandler; |
|||
import org.springframework.amqp.rabbit.annotation.RabbitListener; |
|||
import org.springframework.amqp.support.AmqpHeaders; |
|||
import org.springframework.messaging.handler.annotation.Headers; |
|||
import org.springframework.messaging.handler.annotation.Payload; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
import java.io.IOException; |
|||
import java.util.Date; |
|||
import java.util.Map; |
|||
|
|||
/** |
|||
* @description: |
|||
* @author: wuHuiJuan |
|||
* @create: 2019/12/27 15:40 |
|||
*/ |
|||
@Component |
|||
public class DelayConsumer { |
|||
@RabbitListener(queues = RabbitMQConfig.GAME_STATUS_DELAY_SEND_QUEUE) |
|||
@RabbitHandler |
|||
public void receive(@Payload Object msg, @Headers Map<String, Object> headers, Channel channel ) throws IOException { |
|||
System.out.println("接收到的消息:"+msg +"||" + new Date()); |
|||
|
|||
//ACK 手工签收，通知rabbitMQ，消费端消费成功
|
|||
Long deliveryTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG); |
|||
channel.basicAck(deliveryTag,false); |
|||
} |
|||
|
|||
} |
|||
|
|||
@ -1,53 +1,54 @@ |
|||
//package com.ccsens.util.mq;
|
|||
//
|
|||
//import com.ccsens.util.config.RabbitMQConfig;
|
|||
//import lombok.extern.slf4j.Slf4j;
|
|||
//import org.springframework.amqp.rabbit.connection.CorrelationData;
|
|||
//import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
|||
//import org.springframework.beans.factory.annotation.Autowired;
|
|||
//import org.springframework.stereotype.Component;
|
|||
//
|
|||
///**
|
|||
// * @description:
|
|||
// * @author: wuHuiJuan
|
|||
// * @create: 2019/12/27 15:16
|
|||
// */
|
|||
//@Slf4j
|
|||
//@Component
|
|||
//public class DelayProducer {
|
|||
// @Autowired
|
|||
// private RabbitTemplate rabbitTemplate;
|
|||
//
|
|||
// //消息发送后的回调函数
|
|||
// /**
|
|||
// * 生产者回调函数：confirm确认消息投递成功
|
|||
// */
|
|||
// final RabbitTemplate.ConfirmCallback confirmCallback = (correlationData, ack, cause) -> {
|
|||
//
|
|||
// String messageId = correlationData.getId();
|
|||
// if (ack) {
|
|||
// log.info("消息投递成功，{}",messageId);
|
|||
// //进行消息记录的数据库更新
|
|||
//
|
|||
// }else{
|
|||
// log.info("消息投递失败");
|
|||
// }
|
|||
//
|
|||
// };
|
|||
//
|
|||
// /**
|
|||
// * 通过延迟消息插件发动延迟消息
|
|||
// * @param msg
|
|||
// * @param expiration
|
|||
// */
|
|||
// public void sendDelayMessage(Object msg, Long expiration){
|
|||
//
|
|||
// //绑定异步监听回调函数
|
|||
// rabbitTemplate.setConfirmCallback(confirmCallback);
|
|||
//
|
|||
// rabbitTemplate.convertAndSend(RabbitMQConfig.GAME_STATUS_DELAY_SEND_EXCHANGE,RabbitMQConfig.GAME_STATUS_DELAY_SEND_ROUTING_KEY, msg,(message)->{
|
|||
// message.getMessageProperties().setHeader("x-delay", expiration);//设置延迟时间
|
|||
// return message;
|
|||
// },new CorrelationData("123"));
|
|||
// }
|
|||
//}
|
|||
package com.ccsens.util.mq; |
|||
|
|||
import com.ccsens.util.config.RabbitMQConfig; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.amqp.rabbit.connection.CorrelationData; |
|||
import org.springframework.amqp.rabbit.core.RabbitTemplate; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
/** |
|||
* @description: |
|||
* @author: wuHuiJuan |
|||
* @create: 2019/12/27 15:16 |
|||
*/ |
|||
@Slf4j |
|||
@Component |
|||
public class DelayProducer { |
|||
@Autowired |
|||
private RabbitTemplate rabbitTemplate; |
|||
|
|||
//消息发送后的回调函数
|
|||
/** |
|||
* 生产者回调函数：confirm确认消息投递成功 |
|||
*/ |
|||
final RabbitTemplate.ConfirmCallback confirmCallback = (correlationData, ack, cause) -> { |
|||
|
|||
String messageId = correlationData.getId(); |
|||
if (ack) { |
|||
System.out.println("消息发送成功" + correlationData); |
|||
log.info("消息投递成功，{}",messageId); |
|||
//进行消息记录的数据库更新
|
|||
|
|||
}else{ |
|||
log.info("消息投递失败"); |
|||
} |
|||
|
|||
}; |
|||
|
|||
/** |
|||
* 通过延迟消息插件发动延迟消息 |
|||
* @param msg |
|||
* @param expiration |
|||
*/ |
|||
public void sendDelayMessage(Object msg, Long expiration){ |
|||
|
|||
//绑定异步监听回调函数
|
|||
rabbitTemplate.setConfirmCallback(confirmCallback); |
|||
|
|||
rabbitTemplate.convertAndSend(RabbitMQConfig.GAME_STATUS_DELAY_SEND_EXCHANGE,RabbitMQConfig.GAME_STATUS_DELAY_SEND_ROUTING_KEY, msg,(message)->{ |
|||
message.getMessageProperties().setHeader("x-delay", expiration);//设置延迟时间
|
|||
return message; |
|||
}, new CorrelationData("123")); |
|||
} |
|||
} |
|||
|
|||
Loading…
Reference in new issue