package cc.mrbird.febs.rabbit.producter; import cc.mrbird.febs.mall.dto.OpenPrice; import cc.mrbird.febs.rabbit.constants.ExchangeConstants; import cc.mrbird.febs.rabbit.constants.RouteKeyConstants; import cc.mrbird.febs.rabbit.enumerates.RabbitQueueEnum; import cn.hutool.core.util.IdUtil; import lombok.extern.slf4j.Slf4j; import org.springframework.amqp.AmqpException; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessagePostProcessor; 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; import javax.validation.constraints.NotNull; import java.util.Date; import java.util.UUID; /** * @author wzy * @date 2021-09-25 **/ @Slf4j @Component public class AgentProducer implements RabbitTemplate.ConfirmCallback { /** * 配置中配置的RabbitTemplate的是prototype类型,不能直接注入 */ private RabbitTemplate rabbitTemplate; /** * 在构造方法上注入RabbitTemplate * * @param */ @Autowired public AgentProducer(RabbitTemplate rabbitTemplate) { this.rabbitTemplate = rabbitTemplate; rabbitTemplate.setConfirmCallback(this); } @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { } public void sendChouJiangDelayMsg(String id, Long times) { log.info("发送延时开奖信息消息:{}, {}", id, times); rabbitTemplate.convertAndSend(RabbitQueueEnum.CHOU_JIANG_TIME_TTL.getExchange(), RabbitQueueEnum.CHOU_JIANG_TIME_TTL.getRoute(), id, new MessagePostProcessor() { @Override public Message postProcessMessage(Message message) throws AmqpException { message.getMessageProperties().setExpiration(String.valueOf(times)); return message; } }); } public void sendNoTimeDelayMsg(String id, Long times) { log.info("发送停止下注:{}, {}", id, times); rabbitTemplate.convertAndSend(RabbitQueueEnum.NO_TIME_TTL.getExchange(), RabbitQueueEnum.NO_TIME_TTL.getRoute(), id, new MessagePostProcessor() { @Override public Message postProcessMessage(Message message) throws AmqpException { message.getMessageProperties().setExpiration(String.valueOf(times)); return message; } }); } public void sendOpenPriceMsg(OpenPrice openPrice) { log.info("发送开奖信息:{}, {}", openPrice.getKjNo(), openPrice.getKjNum()); CorrelationData correlationData = new CorrelationData(IdUtil.simpleUUID()); rabbitTemplate.convertAndSend(RabbitQueueEnum.OPEN_PRICE.getExchange(), RabbitQueueEnum.OPEN_PRICE.getRoute(), openPrice, correlationData); } }