注意

概述

快速开始

# 自动安装工程相关的依赖npm install# 初始化 grpc 代码npm run build# 安装rocketmq nodejs 客户端npm i rocketmq-grpc

GRPC_TRACE=compression GRPC_VERBOSITY=debug GRPC_TRACE=all npm run xxx or node xxx

示例

普通消息

import{Producer}from"rocketmq-grpc";constsimpleProducer=newProducer({endpoints:'localhost:8081'});console.log('checkout：simpleProducer init success!');(async()=>{// 启动生产者simpleProducer.startup().then(()=>{console.log('checkout：simpleProducer startup success!');// 发送消息simpleProducer.send({topic:'checkout-topic',tag:'checkout',keys:['checkout-key'],body:Buffer.from('Hello, Checkout OK!')}).then(()=>{console.log('checkout：send message success!');process.exit(0);});});})();

import{SimpleConsumer}from"rocketmq-grpc";constconsumer=newSimpleConsumer({consumerGroup:'checkout-group',endpoints:'192.168.1.162:8081',subscriptions:newMap().set('checkout-topic','*'),requestTimeout:3000,awaitDuration:30000// long polling});console.log('checkout：consumer init success!');constisShutdown=false;asyncfunctionstartAndConsumeMessages(){try{// 启动消费者awaitconsumer.startup();console.log('checkout：consumer startup success!');asyncfunctionconsumeMessages(){try{constmessages=awaitconsumer.receive(20);if(messages.length>0){console.log('got %d messages',messages.length);for(constmessageofmessages){console.log('body=%o',message.body.toString());awaitconsumer.ack(message);console.log('checkout：ack message success!');}}else{console.log('No messages received, waiting...');}}catch(error){console.error('An error occurred:',error);}finally{// // 等待一段时间后递归调用consumeMessages// await new Promise(resolve => setTimeout(resolve, 5000));console.log('checkout：waiting for messages...');if(!isShutdown){awaitconsumeMessages();}}}// 开始消费消息awaitconsumeMessages();}catch(error){console.error('An error occurred:',error);}finally{// 如果发生错误或者接收消息出现问题，可以选择重新启动消费者// 在这里你可以添加相应的逻辑}}startAndConsumeMessages().catch(console.error);

消息类型

消费类型