利用mq实现导入学生库

app.js

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

app/controller/.dataimp.js

@@ -10,7 +10,7 @@ module.exports = {
     parameters: {
       params: ['!id']
     },
-    requestBody: ['!userid', 'name', 'createtime', 'type', 'content', 'status']
+    requestBody: ['!userid', 'name', 'createtime', 'type', 'content', 'status', 'errmsg']
   },
   show: {
     parameters: {

app/controller/home.js

@@ -7,25 +7,6 @@ class HomeController extends Controller {
     const { ctx } = this;
     ctx.body = 'hi, egg';
   }
-
-  // 导入excel
-  async excelimport() {
-    console.log('进入方法');
-    const { ctx } = this;
-    // 获取文件对象
-    console.log(ctx.request.files[0]);
-    const file = ctx.request.files[0];
-    // 将文件解析成js数据，上边封装的可复用的解析函数
-    const userData = await this.service.excelimport.getImportXLSXData(file);
-    // 对解析出来的数据进行校验，如果校验失败，返回错误
-    // TODO
-    console.log(userData);
-    // // 初步校验通过，导入数据库，返回结果
-    // const result = await ctx.service.auth.user.import(userData);
-    // ctx.success({
-    //   data: result,
-    // });
-  }
 }
 
 module.exports = HomeController;

app/model/dataimp.js

@@ -10,6 +10,7 @@ const DataimpSchema = {
   type: { type: String, required: false, maxLength: 64 }, // 类别
   content: { type: String, required: false, maxLength: 64 }, // 内容
   status: { type: String, required: false, maxLength: 2 }, // 状态
+  errmsg: { type: String, required: false, maxLength: 200 }, // 导入错误信息
 };
 
 

app/model/enrollment.js

@@ -0,0 +1,36 @@
+'use strict';
+
+const Schema = require('mongoose').Schema;
+const metaPlugin = require('naf-framework-mongoose/lib/model/meta-plugin');
+
+// 学籍信息信息库
+const SchemaDefine = {
+  schid: { type: String, required: true, maxLength: 64 }, // 学校ID，等同于yxdm
+  year: { type: String, required: true, maxLength: 64 }, // 毕业年份
+  xh: { type: String, required: true, maxLength: 64 }, // 学号
+  xm: { type: String, required: true, maxLength: 64 }, // 姓名
+  sfzh: { type: String, required: true, maxLength: 64 }, // 身份证号
+  xb: { type: String, required: true, maxLength: 64 }, // 性别
+  mz: String, // 民族
+  zzmm: String, // 政治面貌
+  yxdm: { type: String, required: true, maxLength: 64 }, // 院校代码
+  yxmc: { type: String, required: true, maxLength: 64 }, // 院校名称
+  zydm: String, // 专业代码
+  zymc: { type: String, required: true, maxLength: 64 }, // 专业名称
+  xldm: String, // 学历代码
+  xl: { type: String, required: true, maxLength: 64 }, // 学历
+  syszddm: String, // 生源所在地代码
+  syszd: String, // 生源所在地
+  szyx: String, // 所在院系
+  szbj: String, // 所在班级
+};
+const schema = new Schema(SchemaDefine, { toJSON: { virtuals: true } });
+schema.index({ year: 1 });
+schema.index({ year: 1, schid: 1 });
+schema.index({ schid: 1, xh: 1 }, { unique: true, sparse: true }); // 同一学校学号必须唯一
+schema.plugin(metaPlugin);
+
+module.exports = app => {
+  const { mongoose } = app;
+  return mongoose.model('Enrollment', schema, 'stud_enrollment');
+};

app/router.js

@@ -6,8 +6,6 @@
 module.exports = app => {
   const { router, controller } = app;
   router.get('/', controller.home.index);
-  router.post('/api/excelimport', controller.home.excelimport);
-
 
   router.resources('dataimp', '/api/dataimp', controller.dataimp); // index、create、show、destroy
 };

app/schedule/schedule.js

@@ -1,27 +0,0 @@
-'use strict';
-
-const Subscription = require('egg').Subscription;
-let i = 0;
-class Schedule extends Subscription {
-
-  // 通过 schedule 属性来设置定时任务的执行间隔等配置
-  static get schedule() {
-    return {
-      interval: '1m', // 1 分钟间隔
-      type: 'all', // 指定所有的 worker 都需要执行
-    };
-  }
-
-  // subscribe 是真正定时任务执行时被运行的函数
-  async subscribe() {
-    console.log(i++);
-    // 从上传表中取得待读取的文件
-    const data = { status: '0' };
-    const result = await this.ctx.service.dataimp.query(data);
-    console.log(result);
-    // 调用导入方法
-    await this.ctx.service.excelimport.getImportXLSXData(result);
-  }
-}
-
-module.exports = Schedule;

app/service/axiox/student.js

@@ -0,0 +1,20 @@
+'use strict';
+
+const _ = require('lodash');
+const { AxiosService } = require('naf-framework-mongoose/lib/service');
+
+const meta = {
+  create: {
+    uri: '/enrollments',
+    method: 'post',
+  },
+};
+
+class StudentService extends AxiosService {
+  constructor(ctx) {
+    super(ctx, meta, _.get(ctx.app.config, 'axios.stud'));
+    this.model = this.ctx.model.Enrollment;
+  }
+}
+
+module.exports = StudentService;

app/service/dataimp.js

@@ -8,6 +8,29 @@ class DataimpService extends CrudService {
     super(ctx);
     this.model = this.ctx.model.Dataimp;
   }
+
+  async create(requestBody) {
+    console.log(requestBody);
+    // this.ctx.model.create
+    const result = await this.ctx.model.Dataimp.create(requestBody);
+    console.log(result);
+    if (result != null) {
+      const { mq } = this.ctx;
+      if (mq) {
+        const msg = requestBody.name + '上传文件' + requestBody.content;
+        const parm = {
+          durable: true,
+          headers: {
+            userid: requestBody.userid,
+          } };
+        console.log(parm);
+        console.log(msg);
+        await mq.topic('stu_import', requestBody.userid, msg, parm);
+      } else {
+        this.ctx.logger.error('!!!!!!没有配置MQ插件!!!!!!');
+      }
+    }
+  }
 }
 
 module.exports = DataimpService;

app/service/excelimport.js

@@ -12,65 +12,74 @@ class ExcelimportService extends Service {
 
   // 获取导入的XLSX文件中的数据
   async getImportXLSXData(result) {
+    console.log(result);
     for (const elem of result) {
-      const id = elem._id;
+      const _id = elem._id;
       const filepath = this.ctx.app.config.fileDirImp + elem.content;
-      if(elem.type === '0'){
+      if (elem.type === '0') {
         const workbook = XLSX.readFile(filepath);
         // 读取内容
-        let exceldata = [];
-        const sheetNames = workbook.SheetNames; // 获取表名
-        const sheet = workbook.Sheets[sheetNames[0]]; // 通过表名得到表对象
-        const theadRule = [ sheet.A1.v, sheet.B1.v, sheet.C1.v, sheet.D1.v, sheet.E1.v, sheet.F1.v, sheet.G1.v, sheet.H1.v, sheet.I1.v, sheet.J1.v, sheet.K1.v, sheet.L1.v, sheet.M1.v, sheet.N1.v, sheet.O1.v, sheet.P1.v ];
+        const sheetNames = workbook.SheetNames; // 获取所有sheet页
+        const sheet = workbook.Sheets[sheetNames[0]]; // 通过取得当前sheet页
+        // const theadRule = [ sheet.A1.v, sheet.B1.v, sheet.C1.v, sheet.D1.v, sheet.E1.v, sheet.F1.v, sheet.G1.v, sheet.H1.v, sheet.I1.v, sheet.J1.v, sheet.K1.v, sheet.L1.v, sheet.M1.v, sheet.N1.v, sheet.O1.v, sheet.P1.v ];
         const params = XLSX.utils.sheet_to_json(sheet); // 通过工具将表对象的数据读出来并转成json
         // const theadRule = [ '序号', '院系', '班级', '专业代码', '专业名称', '学号', '姓名', '身份证号', '性别', '学历代码', '学历名称', '毕业年份', '民族', '电话号码', '生源所在地', '政治面貌' ];
         console.log(params);
         if (!params) return [];
-        let i = 0;
-        const length = params.length;
-        const _datas = [];
-        let data = {};
-        for (i; i < length; i++) {
-          data = params[i];
-          _datas.push({
-            szyx: data[theadRule[1]],
-            szbj: data[theadRule[2]],
-            zydm: data[theadRule[3]],
-            zymc: data[theadRule[4]],
-            xh: data[theadRule[5]],
-            xm: data[theadRule[6]],
-            sfzh: data[theadRule[7]],
-            xb: data[theadRule[8]],
-            xldm: data[theadRule[9]],
-            xl: data[theadRule[10]],
-            year: data[theadRule[11]],
-            mz: data[theadRule[12]],
-            dhhm: data[theadRule[13]],
-            syszd: data[theadRule[14]],
-            zzmm: data[theadRule[15]],
-          });
-        }
-        exceldata = [ ...exceldata, ..._datas ];
-        console.log(exceldata);
-        // 
-        const stuPath = this.ctx.app.config.baseDirImp + this.ctx.app.config.stusDirImp;
-        const queryData = { year: elem.createtime, schid: elem.userid, schname: elem.name};
-        const studRes = this.ctx.curl(stuPath, {
-          method: 'POST',
-          dataType: 'json',
-          query: queryData,
-          body: exceldata,
+        const maps = {
+          szyx: '院系',
+          szbj: '班级',
+          yxdm: '院校代码',
+          yxmc: '院校名称',
+          zydm: '专业代码',
+          zymc: '专业名称',
+          xh: '学号',
+          xm: '姓名',
+          sfzh: '身份证号',
+          xb: '性别',
+          xldm: '学历代码',
+          xl: '学历名称',
+          year: '毕业年份',
+          mz: '民族',
+          dhhm: '电话号码',
+          syszddm: '生源所在地代码',
+          syszd: '生源所在地',
+          zzmm: '政治面貌',
+        };
+        const _datas = params.map(p => {
+          const data = {};
+          Object.keys(maps).forEach(k => { data[k] = p[maps[k]]; });
+          // Object.entries(maps).forEach(a => { data[a[0]] = p[a[1]]; });
+          return data;
         });
-        console.log(studRes);
+        // exceldata = [ ...exceldata, ..._datas ];
+        // 将数据逐条导入学生库中
+        const queryData = { year: elem.createtime, schid: elem.userid, schname: elem.name };
+        let errCount = 0;
+        const errmsgRes = [];
+        for (const i in _datas) {
+          try {
+            const studRes = await this.ctx.service.axiox.student.create(queryData, _datas[i]);
+            console.log(studRes);
+          } catch (err) {
+            console.log(err);
+            errCount++;
+            errmsgRes.push(_datas[i]);
+          }
+        }
+        console.log(errCount);
         // 导入成功时更新状态
-        if(studRes != null){
-          const updatedata = {id: id, status: '1' };
-          await this.ctx.service.dataimp.update(updatedata);
+        if (errCount === 0) {
+          const updatedata = { status: '1' };
+          await this.ctx.service.dataimp.update({ id: _id }, updatedata);
+        } else {
+          console.log(errmsgRes);
+          const updatedata = { status: '2', errmsg: errmsgRes };
+          console.log(updatedata);
+          await this.ctx.service.dataimp.update({ id: _id }, updatedata);
         }
       }
     }
-    
-    return '0';
   }
 
 }

app/service/rabbitmq.js

@@ -0,0 +1,61 @@
+'use strict';
+
+const Service = require('egg').Service;
+
+class RabbitmqService extends Service {
+
+  constructor(ctx) {
+    super(ctx);
+    this.exType = 'topic';
+    this.durable = true;
+  }
+
+  // 发送消息
+  async sendQueueMsg(ex, routeKey, msg, parm) {
+    const { mq } = this.ctx;
+    if (mq) {
+      await mq.topic(ex, routeKey, msg, parm);
+    } else {
+      this.ctx.logger.error('!!!!!!没有配置MQ插件!!!!!!');
+    }
+  }
+
+  // 接收消息
+  async receiveQueueMsg(ex) {
+    console.log(ex);
+    const self = this;
+    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: true });
+        console.log('==q=', q);
+        // 队列绑定 exchange
+        await ch.bindQueue(q.queue, ex, '*');
+        const getResult = async () => {
+          return new Promise(resolve => {
+            ch.consume(q.queue, msg => {
+              ch.close();
+              resolve(msg);
+            }, { noAck: true });
+          });
+        };
+        const resultMq = await getResult();
+        if (resultMq != null) {
+          const headers = resultMq.properties.headers;
+          const data = { userid: headers.userid, type: '0', status: '0' };
+          const dataimpResult = await this.ctx.service.dataimp.query(data);
+          await this.ctx.service.excelimport.getImportXLSXData(dataimpResult);
+        }
+      } catch (e) {
+        console.log('==e==', e);
+        await ch.close();
+      }
+    } else {
+      this.ctx.logger.error('!!!!!!没有配置MQ插件!!!!!!');
+    }
+  }
+}
+
+module.exports = RabbitmqService;

config/config.default.js

@@ -35,7 +35,7 @@ module.exports = appInfo => {
   // add your config here
   config.cluster = {
     listen: {
-      port: 8106,
+      port: 8107,
     },
   };
 
@@ -61,7 +61,25 @@ module.exports = appInfo => {
   // base路径
   config.baseDirImp = 'http://smart.cc-lotus.info';
   // 学生信息url
-  config.stusDirImp = '/api/enrollments';
+  config.stusDirImp = '/stud/enrollments';
+
+  // axios service config
+  config.axios = {
+    stud: { // 学生服务
+      baseUrl: 'http://localhost:8101/api',
+    },
+  };
+
+  config.amqp = {
+    client: {
+      hostname: '127.0.0.1',
+      username: 'smart',
+      password: 'smart123',
+      vhost: 'smart',
+    },
+    app: true,
+    agent: true,
+  };
 
   // mongoose config
   config.mongoose = {

config/config.local.js

@@ -20,5 +20,17 @@ module.exports = () => {
     },
   };
 
+  // mq config
+  config.amqp = {
+    client: {
+      hostname: '127.0.0.1',
+      username: 'wy',
+      password: '1',
+      vhost: 'smart',
+    },
+    app: true,
+    agent: true,
+  };
+
   return config;
 };

config/plugin.js

@@ -1,9 +1,6 @@
 'use strict';
 
-/** @type Egg.EggPlugin */
-module.exports = {
-  // had enabled by egg
-  // static: {
-  //   enable: true,
-  // }
+exports.amqp = {
+  enable: true,
+  package: 'egg-naf-amqp',
 };

package.json

@@ -8,6 +8,7 @@
   },
   "dependencies": {
     "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",