业务逻辑

一共3个项目.其中一个是和wx相关的.他在收到wx发过来的支付成功消息之后,将信息广播出去,另外两个项目监听这个topic,去做对应的处理.

遇到的问题:

另外2个项目会出现接收不到topic的情况.但是不是一直没有,是偶尔出现. 并且这两个项目丢失的情况不是同一个topic丢失了,而是有的topic一个项目监听到了,另外一个没有收到,每个项目都有大约1/5的几率收不到信息,

这种的不知道该如何下手去处理.

/**
 * activemq
 *
 * [stompit](https://github.com/gdaws/node-stomp)
 */
"use strict";
const {activemq} = require('../config/settings').mq;
const stompit = require('stompit');
const connectOptions = {
 'host' : activemq.host,
 'port' : 61613,
 'connectHeaders': {
 'host' : '/',
 'login' : activemq.user,
 'passcode' : activemq.passwd,
 'heart-beat': '5000,5000'
 }
};
const connectionManager = new stompit.ConnectFailover();
connectionManager.addServer(connectOptions);
// alwarsConnected:
const channel = new stompit.Channel(connectionManager); // 使用这种的会自动重连
// 将任务添加到队列中
exports.publishQueue = publishQueue;
function publishQueue(name, message) {
 const sendHeaders = {
 'destination' : '/queue/' + name,
 'content-type': 'text/plain'
 };
 
 log.silly('添加任务: ', name);
 channel.send(sendHeaders, message, function (err) {
 log.silly('添加任务成功: ', name);
 if (err) {
 log.error('amq -> publishQueue: ' + err)
 }
 });
}
// 发送广播
exports.publishTopic = publishTopic;
function publishTopic(name, message) {
 
 const sendHeaders = {
 'destination' : '/topic/' + name,
 'content-type': 'text/plain'
 };
 
 channel.send(sendHeaders, message, function (err) {
 log.silly('发送广播成功:', name, message);
 if (err) {
 log.error('amq -> publishTopic: ' + err)
 }
 });
}
// 接受广播
exports.subscribeTopic = subscribeTopic;
async function subscribeTopic(name, cb) {
 
 const subscribeHeaders = {
 'destination': '/topic/' + name,
 'ack' : 'client-individual'
 };
 
 channel.subscribe(subscribeHeaders, function (error, message) {
 log.silly('接收广播成功: ', name);
 if (error) {
 return log.error('activemq, subscribeTopic:', error);
 }
 message.readString('utf-8', function (error, body) {
 if (error) {
 return log.error('activemq, subscribeTopic: readString: ', error);
 }
 // channel.ack(message);
 typeof cb === 'function' && cb(body);
 channel.ack(message);
 });
 })
}
// 读取任务
exports.subscribeQueue = subscribeQueue;
async function subscribeQueue(name, cb) {
 const subscribeHeaders = {
 'destination': '/queue/' + name,
 'ack' : 'client-individual'
 };
 
 channel.subscribe(subscribeHeaders, function (error, message) {
 log.silly('读取任务成功: ', name);
 if (error) {
 return log.error('activemq, subscribeQueue:', error, message);
 }
 message.readString('utf-8', function (error, body) {
 if (error) {
 return log.error('activemq, subscribeQueue readString:', error, body);
 }
 // channel.ack(message);
 typeof cb === 'function' && cb(body, channel, message);
 // channel.ack(message); // 这个在每个处理完队列中的任务的时候都必须执行,否则系统会认为这个任务没有执行完成,下次重启的时候会重新读出来,并且堆积的也有数量限制
 });
 })
}