MQ消息队列_发送与接收端工具类

你的名字 2022-04-24 04:06 256阅读 0赞

消息队列 类  
消息队列是异步的，不可能返回对象类型的data数据的，只能返回成功失败的状态

SpoInDBilDlDeaDDDdLineMQProducer.java  
@Component(value =“spoInBDillDeaDdLineMDQProducer”)  
public class SpoInBilDlDeadLinDeMQPDroducer extends BaseMessage \{  
private static final long serialVersionUID = 1L;

public void inBillDeadLineDMessaDgeCreate(String body){
                String topic = "spoInBillDeadLine";
                String tag = "spoInBillDeadLine";
                SimpleMessage message = SimpleMessage.newMessage(topic, tag, body);
                MessageProducerUtils.getInstance().send(message);
        }

SDpoComDmMaDterialMDessDageConsumer.java  
@Component  
@Scope(“singleton”)  
public class SpoCommDMaDteriaDlMesDsageConsumer extends AbstractMessageConsumer\{

private Logger logger = LoggerFactory.getLogger(SDpoComDmMaDterialMDessDageConsumer.class);
    
        @Qualifier("SDpoComDmMaDterialMDessDageConsumer")
    @Autowired
    public DefaultMQPushConsumer defaultMQPushConsumer;
        
        @Autowired
        private  SDpoReturnsOrderMapper  sDpoReturnsOrderMapper;
        
        @PostConstruct
    public void consume() throws MQClientException {
        MessageConsumerConfig config = new MessageConsumerConfig();
        String topic = "spoCommMaterial";
        String tag = "spoCommMaterial";
        config.registTopicTag(topic, tag);
        registConsumer(defaultMQPushConsumer, config);
        start();
        }
        
        @Override
        public ConsumeConcurrentlyStatus process(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                if(CollectionUtils.isNotEmpty(msgs)){
                        logger.info("SDpoCommDMaDterialMessageConsumer_msgs_count_{}",msgs.size());
                        try {
                                for(MessageExt msg:msgs){
                                        String bodyStr = (String) SerializerUtils.deserialize(msg.getBody());
                                        logger.info("SDpoComDmMDaterialDMessageDConsumer_bodyStr_{}",bodyStr);
                                        SpoLog logvo = FastJSONHelper.deserialize(bodyStr, SpoLog.class);
                                    //insertDB
                                        sDpDoReturnsOrderMapper.insertSpoLog(logvo);
                                }
                        } catch (Exception e) {
                                e.printStackTrace();
                        }
                }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    
        @Override
        public void writeErrorMsgToDB(Object arg0, Exception arg1) {
                // TODO Auto-generated method stub
                
        }

使用消息队列  
SDpDDInBDillHeDadBizImpl.java

/\*\*生成入库消息队列，按维度(主体+组织)  
\*/  
private String pushInBilToMQ(String orgCode,List receList)\{  
int count = 0;  
int ibhcount = 0;  
HashMap<String,List> receData = new HashMap<String,List>();  
for(SDDoInBillDeadLineVO rece:receList)\{  
String code = rece.getOrgCode()+"\_"+rece.getUnitCode();  
String inBillNo = rece.getInBillNo();  
if(receData.containsKey(code))\{  
List iblist = receData.get(code);  
if(iblist.contains(inBillNo))\{  
continue;  
\}else\{  
iblist.add(inBillNo);  
ibhcount++;  
\}  
\}else\{  
List iblist = new ArrayList();  
iblist.add(inBillNo);  
receData.put(code, iblist);  
count++;  
ibhcount++;  
\}  
\}  
Set<Entry<String,List>> receSet = receData.entrySet();  
Iterator<Entry<String,List>> receIter = receSet.iterator();  
while(receIter.hasNext())\{  
Entry<String,List> en = receIter.next();  
SDoContactReceiveVO vo = new SDoContactReceiveVO();  
vo.setKey(en.getKey());  
vo.setValue(en.getValue());  
sDInBillDeadDLineMQDProducer.inBillDeadLineMessageCreate(FastJSONHelper.serialize(vo));  
\}  
return “本体:”+orgCode+",单位总数:"+count+",入标单总数:"+ibhcount+",";  
\}

2018/11/10 9:18  
当我使用工作组的mq的182 端口号容易消耗别人的消息，导致别人接受不到消息。后改成181，这样是不会消费本项目组的消息，但是会消耗别人的mq消息，导致控制台输出。这样容易冲掉本来是我要观察的sql。后来又改mq消费端口成189 ， 结果rpc启动就是这样子，在这段日志输出前就是20000毫秒正常。但是就是等待这段代码输出花费40000多毫秒，后经过分析 是mq一直在等待接通，后来一直接通不了，才启动整个项目完成。

2018-11-09 20:33:05.611 DEBUG \[localhost-startStop-1\] i.n.channel.ChannelOutboundBuffer \[Slf4JLogger.java : 76\] - -Dio.netty.threadLocalDirectBufferSize: 65536  
2018-11-09 20:33:05.631 DEBUG \[localhost-startStop-1\] io.netty.buffer.ByteBufUtil \[Slf4JLogger.java : 76\] - -Dio.netty.allocator.type: unpooled  
2018-11-09 20:33:05.640 DEBUG \[nioEventLoopGroup-2-1\] i.n.u.i.JavassistTypeParameterMatcherGenerator \[Slf4JLogger.java : 76\] - Generated: io.netty.util.internal.**matchers**.com.alibaba.rocketmq.remoting.protocol.RemotingCommandMatcher  
2018-11-09 20:33:14.693 INFO \[localhost-startStop-1\] c.m.c.c.c.m.r.CreateServiceOrderConsumer \[CreateServiceOrderConsumer.java : 133\] - CreateServiceOrderConsumer start,listening on 10.16.68.189:9876  
2018-11-09 20:33:24.242 INFO \[localhost-startStop-1\] c.m.c.c.c.m.r.MdfDistributePaySuccessConsumer \[MdfDistributePaySuccessConsumer.java : 82\] - MdfDistributePaySuccessConsumer start,listening on 10.16.68.189:9876  
2018-11-09 20:33:33.708 INFO \[localhost-startStop-1\] c.m.c.c.c.m.r.PaySuccessConsumer \[PaySuccessConsumer.java : 106\] - PaySuccessConsumer start,listening on 10.16.68.189:9876  
2018-11-09 20:33:43.224 INFO \[localhost-startStop-1\] c.m.c.c.c.m.r.RefundSuccessConsumer \[RefundSuccessConsumer.java : 90\] - RefundSuccessConsumer start,listening on 10.16.68.189:9876  
2018-11-09 20:33:52.670 INFO \[localhost-startStop-1\] c.m.c.c.c.m.r.ServiceOrderStatusConsumer \[ServiceOrderStatusConsumer.java : 175\] - ServiceOrder-Status-Consumer Started,listen on server:10.16.68.189:9876  
十一月 09, 2018 8:33:54 下午 org.apache.coyote.AbstractProtocol start  
信息: Starting ProtocolHandler \[“http-bio-8380”\]  
十一月 09, 2018 8:33:54 下午 org.apache.coyote.AbstractProtocol start  
信息: Starting ProtocolHandler \[“ajp-bio-8309”\]  
十一月 09, 2018 8:33:54 下午 org.apache.catalina.startup.Catalina start  
信息: Server startup in 68776 ms