hh

app.js

@@ -0,0 +1,28 @@
+'use strict';
+class AppBootHook {
+  constructor(app) {
+    this.app = app;
+  }
+
+  async didReady() {
+    // 应用已经启动完毕
+    const ctx = await this.app.createAnonymousContext();
+    // 企业入驻申请消息接收事件
+    await ctx.service.rabbitmq.receiveQueueMsg('public_chat');
+  }
+
+  async serverDidReady() {
+    // 应用已经启动完毕
+  }
+}
+module.exports = AppBootHook;
+
+// module.exports = app => {
+//   app.messenger.on('mq_action', data => {
+//     console.log('------------------------------->app-start');
+//     // 应用已经启动完毕
+//     const ctx = app.createAnonymousContext();
+//     // 企业入驻申请消息接收事件
+//     ctx.service.rabbitmq.receiveQueueMsg('stu_import');
+//   });
+// };

app/service/chat.js

@@ -20,7 +20,12 @@ class ChatService extends CrudService {
     const { mq } = this.ctx;
     if (mq) {
       const exchange = 'public_chat';
-      await mq.topic(exchange, '', res.data, '');
+      const parm = {
+        durable: true,
+        headers: {
+          userid: 1,
+        } };
+      await mq.topic(exchange, '1', content, parm);
     }
     return res;
   }

app/service/rabbitmq.js

@@ -0,0 +1,39 @@
+'use strict';
+
+const Service = require('egg').Service;
+
+class RabbitmqService extends Service {
+
+  constructor(ctx) {
+    super(ctx);
+    this.exType = 'topic';
+    this.durable = true;
+  }
+
+  // 接收消息
+  async receiveQueueMsg(ex) {
+    console.log(ex);
+    this.ctx.logger.info('调用mq的' + ex);
+    const self = this;
+    const { mq } = self.ctx;
+    if (mq) {
+      const ch = await mq.conn.createChannel();
+      await ch.assertExchange(ex, 'topic', { durable: true });
+      const q = await ch.assertQueue('', { exclusive: true });
+      await ch.bindQueue(q.queue, ex, '*');
+      await ch.consume(q.queue, msg => this.logMessage(msg, this), { autoAck: true });
+    } else {
+      this.ctx.logger.error('!!!!!!没有配置MQ插件!!!!!!');
+    }
+  }
+
+  async logMessage(msg) {
+
+    const result = msg.content.toString();
+    console.log(result);
+
+  }
+
+}
+
+module.exports = RabbitmqService;