hhhh

app/controller/home.js

@@ -1,5 +1,7 @@
 'use strict';
 
+const amqp = require('amqplib');
+const msgValue = require('../util/constants');
 const Controller = require('egg').Controller;
 
 class HomeController extends Controller {
@@ -7,6 +9,22 @@ class HomeController extends Controller {
     const { ctx } = this;
     ctx.body = 'hi, egg';
   }
+
+  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)));
+    ctx.body = '发送成功';
+  }
+
+  async receivemq() {
+    const { ctx, app } = this;
+    await this.service.rabbitmq.receiveQueueMsg(ctx.query.exchange,ctx.query.routekey,(msg) => 
+    {    
+       console.log(msg);
+    });
+    ctx.body = "接收成功";
+  }
 }
 
 module.exports = HomeController;

app/router.js

@@ -6,6 +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.resources('message', '/api/message', controller.message); // index、create、show、destroy
 

app/service/rabbitmq.js

@@ -0,0 +1,79 @@
+/* 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 {
+  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) {
+    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('发送成功');
+      } else {
+        console.log('发送失败');
+      }
+      await ch.close();
+    } catch (e) {
+      console.log('==e==', e);
+      await ch.close();
+    }
+  }
+
+  // 接收消息
+  async receiveQueueMsg(queueName, 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();
+    }
+  }
+
+}
+
+module.exports = RabbitmqnewService;

app/util/constants.js

@@ -1,7 +1,10 @@
 'use strict';
 
-exports.MsgTypeValues = {
-  STU: '0', // 待审核
-  PASS: '1', // 审核通过
-  REJECT: '2', // 审核失败
+exports.MsgValues = {
+  STU: '0', // 学生
+  CROP: '1', // 企业
+  SCH: '2', // 学校
+  EXCHANGE_CROP_REG: 'crop_register', // 企业注册
+  EXCHANGE_CROP_APPLY: 'crop_apply', // 企业申请三会
+  EXCHANGE_STU_APPLY: 'stu_apply', // 学生投递简历
 };

package.json

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