• nestjs使用rabbitMQ

    基础知识

    rabbitMQ的exchange、bindingkey、routingkey的关系：

    Java面试常见问题：RabbitMQ基本概念与工作原理 - 知乎

    nestjs的mq使用引导（最后没按照官网的那种方式使用，而是直接编程的方式，这里看下了解下）：

    RabbitMQ | NestJS 中文文档 | NestJS 中文网

    其中，使用的mqlib的API（非常重要，查看了官方的API才能更精准的使用！）：

    amqplib | Channel API reference

    软件

    nestjs 8.*.*

    使用rabbitMQ需要增加依赖（为什么？看官网：）：

    "amqp-connection-manager": "^4.1.4",
"amqplib": "^0.10.2",

    怎么安装? 不想全局就在当前项目的根目录下执行：

    npm install --save amqp-connection-manager amqplib

    然后，开发，上代码

    代码

    mt.tuil.ts

    1. import {Injectable, Logger, OnModuleInit} from '@nestjs/common';
    2. import amqp from 'amqplib';
    3. import * as acm from 'amqp-connection-manager';
    4. import {MqMsgHandler} from './MQMsgHandler';
    5. import {GLoggerService} from '../../framework/logs/glogger';
    6. import ChannelWrapper from "amqp-connection-manager/dist/esm/ChannelWrapper";
    7.  
    8. const logger = new GLoggerService('RabbitMQ');
    9.  
    10. // type OnEventHandler = (data: any) => Promise;
    11.  
    12. // @Injectable()
    13. export class RabbitMQClient {
    14.     private connection: acm.AmqpConnectionManager;
    15.     private channelWrapper: acm.ChannelWrapper;
    16.     private rabbitMqConfig: {};
    17.     private url: string;
    18.     private exchange: string;// = 'exchange_event_consume';
    19.     private bindKey: string;
    20.     private virtualHost: string;
    21.     private host: string;
    22.     private port: number;
    23.     private username: string;
    24.     private password: string;
    25.     private msgHandler: MqMsgHandler | undefined;// 是否添加监听者，用于消费使用
    26.     private queue: string;
    27.  
    28.     constructor(rabbitMqConfig, msgHandler?: MqMsgHandler) {
    29.         this.rabbitMqConfig = rabbitMqConfig;
    30.         this.virtualHost = rabbitMqConfig.virtualHost;
    31.         this.exchange = rabbitMqConfig.exchange;
    32.         this.bindKey = rabbitMqConfig.bindKey;
    33.         this.host = rabbitMqConfig.host;
    34.         this.port = rabbitMqConfig.port;
    35.         this.username = rabbitMqConfig.username;
    36.         this.password = rabbitMqConfig.password;
    37.         this.msgHandler = msgHandler;
    38.         this.queue = rabbitMqConfig.queue;
    39.         this.url = 'amqp://' + rabbitMqConfig.username + ':' + rabbitMqConfig.password + '@'
    40.             + rabbitMqConfig.host + ':' + rabbitMqConfig.port + '/%2F' + rabbitMqConfig.virtualHost;
    41.     }
    42.  
    43.     public async connect():Promise<void> {
    44.         return new Promise<void>((resolve, reject) => {
    45.             this.connection = acm.connect(this.url, {
    46.                 heartbeatIntervalInSeconds: 3,
    47.                 reconnectTimeInSeconds: 3,
    48.                 connectionOptions: {
    49.                     rejectUnauthorized: false
    50.                 },
    51.  
    52.             });
    53.  
    54.             this.connection.addListener('connect', () => {
    55.                 logger.debug('[客户端]连接成功');
    56.                 return resolve();
    57.             });
    58.             this.connection.addListener('connectFailed', (err) => {
    59.                 logger.error('[客户端]连接失败！', err.toString());
    60.                 return resolve();
    61.             });
    62.             this.connection.addListener('disconnect', (err) => {
    63.                 logger.log('[客户端]断开连接%O');
    64.                 return reject();
    65.             });
    66.         });
    67.     }
    68.  
    69.     /**
    70.      * 初始化通道
    71.      */
    72.     async initChannel():Promise<ChannelWrapper> {
    73.         const bindKey = this.bindKey;
    74.         const exchange = this.exchange;
    75.         let channelWrapper;
    76.         const queueName = this.queue;
    77.         if (this.msgHandler) {
    78.             const msgHandler = this.msgHandler;
    79.             logger.log('初始化通道[exchange=' + exchange + ', bindKey=' + bindKey + ', queue=' + queueName + ']，并增加通道监听器...');
    80.             channelWrapper = this.connection.createChannel({
    81.                 json: false,
    82.                 setup: async (channel: amqp.Channel): Promise<any> => {
    83.                     // 初始化事件Exchange
    84.                     await channel.assertExchange(exchange, 'direct', {durable: true});
    85.                     // 初始化事件队列
    86.                     await channel.assertQueue(queueName, {
    87.                         durable: false,
    88.                         autoDelete: true
    89.                     });
    90.                     // 绑定事件队列
    91.                     await channel.bindQueue(queueName, exchange, bindKey);
    92.  
    93.                     // 设置预消费
    94.                     await channel.prefetch(1);
    95.                     // 消费事件
    96.                     // tslint:disable-next-line:only-arrow-functions
    97.                     channel.consume(queueName, async function (msg: amqp.ConsumeMessage | null): Promise<void> {
    98.                         let data: any;
    99.                         try {
    100.                             if (msg !== undefined && msg !== null && msg.content !== undefined && msg.content !== null) {
    101.                                 data = msg.content.toString('utf8') as any;
    102.                                 data = JSON.parse(data);
    103.                                 if (msgHandler !== undefined && msgHandler !== null) {
    104.                                     await msgHandler(data);
    105.                                 } else {
    106.                                     logger.debug(data)
    107.                                 }
    108.                                 channel.ack(msg);// 应答
    109.                             } else {
    110.                                 logger.debug('事件数据异常 ' + data);
    111.                                 return Promise.reject(`事件数据异常`);
    112.                             }
    113.                         } catch (error) {
    114.                             logger.debug('[consume]' + error);
    115.                             channel.ack(msg);
    116.                         }
    117.  
    118.  
    119.                     }, {noAck: false});// 需要应答
    120.                 }
    121.             });
    122.             logger.log('完成!');
    123.         } else {
    124.             logger.log('初始化通道[exchange=' + exchange + ', bindKey=' + bindKey + ', queue=' + queueName + ']');
    125.             channelWrapper = this.connection.createChannel({
    126.                 json: false
    127.             });
    128.             logger.log('完成!');
    129.         }
    130.  
    131.         channelWrapper.on('connect', () => {
    132.             logger.log('[通道]连接成功');
    133.         });
    134.         channelWrapper.on('close', () => {
    135.             logger.log('[通道]连接关闭');
    136.         });
    137.         channelWrapper.on('error', (err) => {
    138.             logger.error('[通道]连接错误: ', err.toString());
    139.         });
    140.         this.channelWrapper = channelWrapper;
    141.         return channelWrapper;
    142.     }
    143.  
    144.     // tslint:disable-next-line:no-unused-expression
    145.     async publish(data: any | {}): Promise<boolean> {
    146.         if (!this.channelWrapper) {
    147.             logger.error('通道没有初始化！', '');
    148.             return Promise.reject('通道没有初始化！');
    149.         }
    150.         const BIND_KEY = this.bindKey;
    151.         const exchange = this.exchange;
    152.         return this.channelWrapper
    153.             .publish(exchange, BIND_KEY, Buffer.from(JSON.stringify(data)))
    154.             // .sendToQueue(queue, Buffer.from(JSON.stringify(data)))
    155.             .then(() => {
    156.                 logger.log('publish Message was sent!')
    157.                 return true;
    158.             })
    159.             .catch((err) => {
    160.                 logger.error('publish Message was rejected...  Boo!', err.toString());
    161.                 return false;
    162.             });
    163.     }
    164.  
    165.  
    166.     // tslint:disable-next-line:no-unused-expression
    167.     public async start() {
    168.         await this.connect();
    169.         await this.initChannel();
    170.     }
    171. }

    说明：

    1. 我封装的这个工具用来创建一个客户端出来，通过publish来发布数据，通过创建之初执行初始化方法start()来初始化连接（没交给nest的inject是因为我需要具备自己的特殊控制）。

    2. 创建时构造函数传入参数，第一个参数是初始化参数，结构见rabbitconfig.ts.第二个参数是订阅的回调函数，用于处理收听到的消息，如果不传就不会创建监听。这个回调函数定义了一个类型，见mqMsgHandler.ts.

    mqmsghandler.ts

    export type MqMsgHandler = (data: any) => Promise<boolean>;

    说明：消息的回调函数

    rabbitconfig.ts

    1. export class RabbitConfig{
    2.     private _host:string;
    3.     private _port:number;
    4.     private _queue:string;
    5.     private _bindKey:string;
    6.     private _exchange:string;
    7.     private _username:string;
    8.     private _password:string;
    9.     private _virtualHost:string;
    10.  
    11.  
    12.     constructor(host: string, port: number, queue: string, bindKey: string, exchange: string, username: string, password: string, virtualHost: string) {
    13.         this._host = host;
    14.         this._port = port;
    15.         this._queue = queue;
    16.         this._bindKey = bindKey;
    17.         this._exchange = exchange;
    18.         this._username = username;
    19.         this._password = password;
    20.         this._virtualHost = virtualHost;
    21.     }
    22.  
    23.     get host(): string {
    24.         return this._host;
    25.     }
    26.  
    27.     set host(value: string) {
    28.         this._host = value;
    29.     }
    30.  
    31.     get port(): number {
    32.         return this._port;
    33.     }
    34.  
    35.     set port(value: number) {
    36.         this._port = value;
    37.     }
    38.  
    39.     get queue(): string {
    40.         return this._queue;
    41.     }
    42.  
    43.     set queue(value: string) {
    44.         this._queue = value;
    45.     }
    46.  
    47.     get bindKey(): string {
    48.         return this._bindKey;
    49.     }
    50.  
    51.     set bindKey(value: string) {
    52.         this._bindKey = value;
    53.     }
    54.  
    55.     get exchange(): string {
    56.         return this._exchange;
    57.     }
    58.  
    59.     set exchange(value: string) {
    60.         this._exchange = value;
    61.     }
    62.  
    63.     get username(): string {
    64.         return this._username;
    65.     }
    66.  
    67.     set username(value: string) {
    68.         this._username = value;
    69.     }
    70.  
    71.     get password(): string {
    72.         return this._password;
    73.     }
    74.  
    75.     set password(value: string) {
    76.         this._password = value;
    77.     }
    78.  
    79.     get virtualHost(): string {
    80.         return this._virtualHost;
    81.     }
    82.  
    83.     set virtualHost(value: string) {
    84.         this._virtualHost = value;
    85.     }
    86. }

    net_disk.service.ts

    注意：代码经过一些删除处理，可能直接粘贴无法运行，但是核心逻辑都在里面。

    1. import {GLoggerService} from '../framework/logs/glogger';
    2. import {RabbitMQClient} from '../util/mq/mq.util';
    3. import {Injectable} from '@nestjs/common';
    4. import * as config from 'config';
    5. import {RabbitConfig} from '../util/mq/RabbitConfig';
    6. import {MqMsgHandler} from '../util/mq/MQMsgHandler';
    7. import {data} from 'cheerio/lib/api/attributes';
    8.  
    9. const rabbitMQ = config['rabbitMQ'];
    10.  
    11. @Injectable()
    12. export class NetDiskService {
    13.     private readonly glogger = new GLoggerService();
    14.     private mQClientSubscriber: RabbitMQClient;//订阅者
    15.     private mQClientSender: RabbitMQClient;//发布者
    16.  
    17.     /**
    18.      * 由于定于与发布不是同一个queue，所以需要建立两个client
    19.      */
    20.     constructor() {
    21.         const subscribe = new RabbitConfig(
    22.             rabbitMQ.host,
    23.             rabbitMQ.port,
    24.             rabbitMQ.subscribe.queue,
    25.             rabbitMQ.subscribe.bindKey,
    26.             rabbitMQ.exchange,
    27.             rabbitMQ.username,
    28.             rabbitMQ.password,
    29.             rabbitMQ.virtualHost
    30.         );
    31.         this.mQClientSubscriber = new RabbitMQClient(subscribe,
    32.             // tslint:disable-next-line:no-empty
    33.             async (msg: any) => {
    34.                 console.log('收到0发送的消息:' + JSON.stringify(msg));
    35.                 return await this.onNetDiskUploadGeomFile(msg);
    36.             }
    37.         );
    38.  
    39.         const send = new RabbitConfig(
    40.             rabbitMQ.host,
    41.             rabbitMQ.port,
    42.             rabbitMQ.send.queue,
    43.             rabbitMQ.send.bindKey,
    44.             rabbitMQ.exchange,
    45.             rabbitMQ.username,
    46.             rabbitMQ.password,
    47.             rabbitMQ.virtualHost
    48.         );
    49.         this.mQClientSender = new RabbitMQClient(send,async (msg: any) => {
    50.             console.log('收到1发送的消息:' + JSON.stringify(msg));
    51.             return Promise.resolve(true);
    52.         });
    53.  
    54.         this.mQClientSubscriber.start();
    55.         this.mQClientSender.start();
    56.     }
    57.  
    58.     
    59.     public async publishMTo1(metaData: any): Promise<number> {
    60.         try {
    61.             console.log("发送消息给1")
    62.             let result = await this.mQClientSender.publish(metaData);
    63.             if (!result) {
    64.                 return 1
    65.             }
    66.             return 0
    67.         } catch (e) {
    68.             this.glogger.error('失败！', e);
    69.         }
    70.  
    71.         return 1
    72.     }
    73.  
    74.  
    75.     
    76.     public async publishTo0(metaData: any): Promise<number> {
    77.         try {
    78.             console.log("发送消息给0")
    79.             let result = await this.mQClientSubscriber.publish(metaData);
    80.             if (!result) {
    81.                 return 1
    82.             }
    83.             return 0
    84.         } catch (e) {
    85.             this.glogger.error('失败！', e);
    86.         }
    87.  
    88.         return 1
    89.     }
    90. }

    完成

  • 相关阅读:
    java-php-python-springboot网上订餐系统计算机毕业设计
    【LeetCode】611.有效三角形的个数
    CentOS Stream 9 配置静态 IP
    [附源码]Python计算机毕业设计java高校社团管理系统
    学校介绍静态HTML网页设计作品 DIV布局学校官网模板代码 DW大学网站制作成品下载 HTML5期末大作业
    电脑小白也能修电脑？这几个维修技巧快记住
    敏捷12原则
    Django 用户认证
    EthernetIP 转MODBUS RTU协议网关连接FANUC机器人作为EthernetIP通信从站
    HIVE伪分布安装
  • 原文地址：https://blog.csdn.net/jfqqqqq/article/details/126281403