使用mq插件

app/controller/home.js

@@ -1,7 +1,5 @@
 'use strict';
 
-const amqp = require('amqplib');
-const msgValue = require('../util/constants');
 const Controller = require('egg').Controller;
 
 class HomeController extends Controller {
@@ -11,22 +9,18 @@ class HomeController extends Controller {
   }
 
   async sendmq() {
-    const { ctx, app } = this;
-    console.log(msgValue.MsgValues.EXCHANGE_CROP_REG);
-    await this.service.rabbitmq.sendQueueMsg(ctx.query.exchange, ctx.query.routekey, new Buffer(JSON.stringify(ctx.query.msg)), data => {
-      console.log(data);
-
-    });
-    ctx.body = '发送成功';
+    const { ctx } = this;
+    await this.service.rabbitmq.sendQueueMsg(ctx.query.ex, ctx.query.routekey, ctx.query.msg);
+    this.ctx.body = '发送成功';
   }
 
   async receivemq() {
-    const { ctx, app } = this;
-    await this.service.rabbitmq.receiveQueueMsg(ctx.query.exchange, ctx.query.routekey, msg => {
+    const { ctx } = this;
+    await this.service.rabbitmq.receiveQueueMsg(ctx.query.ex, ctx.query.routekey, msg => {
       console.log(msg);
 
       // 插入待办事项到数据库中。
-      this.service.message.create({ producerid: msg.producerid, consumerid: msg.consumerid, type: msg.type, content: msg.content, remark: msg.remark });
+      this.service.message.create({ userid: msg.userid, name: msg.name, createtime: msg.createtime, type: msg.type, content: msg.content, remark: msg.remark });
 
     });
     ctx.body = '接收成功';

app/model/message.js

@@ -4,17 +4,18 @@ const metaPlugin = require('naf-framework-mongoose/lib/model/meta-plugin');
 
 // 待办事项表
 const MessageSchema = {
-  producerid: { type: String, required: true, maxLength: 64 }, // 生产者
-  consumerid: { type: String, required: false, maxLength: 128 }, // 消费者
+  userid: { type: String, required: true, maxLength: 64 }, // 人员
+  name: { type: String, required: false, maxLength: 200 }, // 名称
+  createtime: { type: String, required: false, maxLength: 20 }, // 时间
   type: { type: String, required: false, maxLength: 64 }, // 类别
   content: { type: String, required: false, maxLength: 64 }, // 内容
+  status: { type: String, required: false, maxLength: 2 }, // 状态
   remark: { type: String, maxLength: 128 }, // 备注
 };
 
 
 const schema = new Schema(MessageSchema, { toJSON: { virtuals: true } });
-schema.index({ producerid: 1 });
-schema.index({ consumerid: 1 });
+schema.index({ userid: 1 });
 schema.plugin(metaPlugin);
 
 module.exports = app => {

app/model/msg.js

@@ -1,29 +0,0 @@
-/**
- * 求职信
- */
-'use strict';
-const Schema = require('mongoose').Schema;
-const metaPlugin = require('naf-framework-mongoose/lib/model/meta-plugin');
-
-// 求职信信息
-const SchemaDefine = {
-  post_id: { type: String, require: true }, // 职位ID
-  resume_id: { type: String, require: true }, // 简历ID(学生ID)
-  studname: { type: String, required: true, maxLength: 64 }, // 学生姓名
-  corpname: { type: String, required: true, maxLength: 128 }, // 企业名称
-  title: { type: String, required: true, maxLength: 128 }, // 职位标题
-  status: { type: String, default: '0' }, // 状态: 0-新投递，未接收；1-已接收；2-已回绝
-  type: { type: String, default: '0' }, // 求职信类型：0-在线招聘；1-招聘会；2-宣讲会；
-  origin: { type: String, require: false }, // 信息来源ID：招聘信息ID、招聘会ID
-  remark: { type: String, required: false, maxLength: 256 },
-};
-const schema = new Schema(SchemaDefine, { toJSON: { virtuals: true } });
-schema.index({ post_id: 1 });
-schema.index({ resume_id: 1 });
-schema.index({ post_id: 1, resume_id: 1 }, { unique: true });
-schema.plugin(metaPlugin);
-
-module.exports = app => {
-  const { mongoose } = app;
-  return mongoose.model('JobsMsg', schema, 'jobs_msg');
-};

app/router.js

@@ -6,8 +6,8 @@
 module.exports = app => {
   const { router, controller } = app;
   router.get('/', controller.home.index);
-  router.get('/api/sendmq',controller.home.sendmq);
-  router.get('/api/receivemq',controller.home.receivemq);
+  router.get('/api/sendmq', controller.home.sendmq);
+  router.get('/api/receivemq', controller.home.receivemq);
   // 待办事项
   router.resources('message', '/api/message', controller.message); // index、create、show、destroy
 

app/service/rabbitmq.js

@@ -1,81 +1,54 @@
-/* eslint-disable no-mixed-spaces-and-tabs */
 'use strict';
 
 const _ = require('lodash');
 const Service = require('egg').Service;
-const amqp = require('amqplib');
 
-class RabbitmqnewService extends Service {
+class RabbitmqService extends Service {
+
   constructor(ctx) {
     super(ctx);
-
-    this.hosts = [{
-      hostname: '127.0.0.1',
-      port: '5672',
-      username: 'wy',
-      password: '1',
-      authMechanism: 'AMQPLAIN',
-      pathname: '/',
-      ssl: {
-        enabled: false,
-      },
-    }];
-    this.index = 0;
     this.exType = 'topic';
     this.durable = true;
-    this.autoDelete = true;
   }
 
   // 发送消息
-  async sendQueueMsg(queueName, routeKey, msg, sendCallBack) {
+  async sendQueueMsg(ex, routeKey, msg) {
     const self = this;
-    const conn = await amqp.connect(self.hosts[self.index]);
-    const ch = await conn.createConfirmChannel();
-    try {
-      await ch.assertExchange(queueName, this.exType, { durable: this.durable });
-      const result = await ch.publish(queueName, routeKey, Buffer.from(msg), {
-        persistent: true, // 消息持久化
-        mandatory: true,
-      });
-      console.log('==result==', result);
-      if (result) {
-        console.log('发送成功');
-        sendCallBack && sendCallBack(true);
-      } else {
-        console.log('发送失败');
-        sendCallBack && sendCallBack(false);
-      }
-      await ch.close();
-    } catch (e) {
-      console.log('==e==', e);
-      await ch.close();
+    const { mq } = self.ctx;
+    if (mq) {
+      await mq.topic(ex, msg, routeKey, { durable: self.durable });
+    } else {
+      this.ctx.logger.error('!!!!!!没有配置MQ插件!!!!!!');
     }
   }
 
   // 接收消息
-  async receiveQueueMsg(queueName, routeKey, receiveCallBack) {
+  async receiveQueueMsg(ex, routeKey, receiveCallBack) {
     const self = this;
-    const conn = await amqp.connect(self.hosts[self.index]);
-    const ch = await conn.createConfirmChannel();
-    try {
-      await ch.assertExchange(queueName, this.exType, { durable: this.durable });
-      const q = await ch.assertQueue('', { exclusive: false });
-      console.log('==q=', q);
-      // 队列绑定 exchange
-      await ch.bindQueue(q.queue, queueName, routeKey);
-      await ch.consume(q.queue, msg => {
-        console.log('收到消息: ', msg);
-        const data = msg.content.toString();
-        // 发送确认消息
-        ch.ack(msg);
-        receiveCallBack && receiveCallBack(data);
-      }, { noAck: false });
-    } catch (e) {
-      console.log('==e==', e);
-      await ch.close();
+    const { mq } = self.ctx;
+    if (mq) {
+      const ch = await mq.conn.createChannel();
+      try {
+        await ch.assertExchange(ex, self.exType, { durable: self.durable });
+        const q = await ch.assertQueue('', { exclusive: false });
+        console.log('==q=', q);
+        // 队列绑定 exchange
+        await ch.bindQueue(q.queue, ex, routeKey);
+        await ch.consume(q.queue, msg => {
+          console.log('收到消息: ', msg);
+          // 发送确认消息
+          ch.ack(msg);
+          receiveCallBack && receiveCallBack(msg);
+        }, { noAck: false });
+      } catch (e) {
+        console.log('==e==', e);
+        await ch.close();
+      }
+    } else {
+      this.ctx.logger.error('!!!!!!没有配置MQ插件!!!!!!');
     }
   }
 
 }
 
-module.exports = RabbitmqnewService;
+module.exports = RabbitmqService;

config/config.default.js

@@ -50,6 +50,19 @@ module.exports = appInfo => {
   //     useCreateIndex: true,
   //   },
   // };
+
+  // mq config
+  config.amqp = {
+    client: {
+      hostname: '127.0.0.1',
+      username: 'wy',
+      password: '1',
+      vhost: 'smart',
+    },
+    app: true,
+    agent: true,
+  };
+
   config.mongoose = {
     url: 'mongodb://localhost:27017/smart',
     options: {

config/plugin.js

@@ -3,3 +3,8 @@
 exports.multiTenancy = {
   enable: false,
 };
+
+exports.amqp = {
+  enable: true,
+  package: 'egg-naf-amqp',
+};

package.json

@@ -7,8 +7,8 @@
     "framework": "naf-framework-mongoose"
   },
   "dependencies": {
-    "amqplib": "^0.5.5",
     "egg": "^2.23.0",
+    "egg-naf-amqp": "0.0.13",
     "egg-scripts": "^2.11.0",
     "jsonwebtoken": "^8.5.1",
     "naf-framework-mongoose": "^0.6.11"