网站免费建站众享星球,wordpress转发插件,linux有wordpress,文章网站哪里建设好【RocketMQ入门-安装部署与Java API测试】 一、环境说明二、安装部署三、Java API 编写Producer和Consumer进行测试四、小结 一、环境说明
虚拟机VWMare#xff1a;安装centos7.6操作系统源码包#xff1a;rocketmq-all-5.1.3-source-release.zip单master部署#xff0c;在… 【RocketMQ入门-安装部署与Java API测试】 一、环境说明二、安装部署三、Java API 编写Producer和Consumer进行测试四、小结 一、环境说明
虚拟机VWMare安装centos7.6操作系统源码包rocketmq-all-5.1.3-source-release.zip单master部署在一台虚拟机上安装部署name server和proxy以及broker流程图
二、安装部署 源码包安装需要事先安装部署maven下载apache-maven-3.6.3-bin.tar.gz安装包然后解压并配置环境变量如下命令 tar -zvxf apache-maven-3.6.3-bin.tar.gz -C /training/配置环境变量此处是用root安装编辑vi ~/.bash_profile在文件末尾添加如下内容 #maven
export MVN_HOME/training/apache-maven-3.6.3
export PATH$MVN_HOME/bin:$PATH执行source ~/.bash_profile 使环境生效。 进入/training/apache-maven-3.6.3/conf目录下配置maven的仓库为阿里云和华为云仓库执行如下命令 cd /training/apache-maven-3.6.3/conf/
mv settings.xml settings.xml.backup
vi settings.xml在打开的settings.xml中粘贴如下内容即可 ?xml version1.0 encodingutf-8?
settings xmlnshttp://maven.apache.org/SETTINGS/1.0.0xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocation http://maven.apache.org/SETTINGS/1.0.0 http://maven.apache.org/xsd/settings-1.0.0.xsdmirrorsmirroridaliyunmaven/idmirrorOf*/mirrorOfname阿里云公共仓库/nameurlhttps://maven.aliyun.com/repository/public/url/mirrormirroridhuaweicloud/idmirrorOfcentral/mirrorOfnamehuaweicloud maven/nameurlhttps://mirrors.huaweicloud.com/repository/maven//url/mirror/mirrorsprofilesprofilerepositoriesrepositoryidcentral/idurlhttps://maven.aliyun.com/repository/central/urlreleasesenabledtrue/enabled/releasessnapshotsenabledtrue/enabled/snapshots/repository/repositories/profile/profiles
/settings由于CentOS7.6最小模式安装没有unzip命令需要事先安装执行如下命令安装 yum install unzip -y解压源码包rocketmq-all-5.1.3-source-release.zip进入到解压后的目录下然后编译安装执行如下命令 unzip rocketmq-all-5.1.3-source-release.zip
cd rocketmq-all-5.1.3-source-release/
mvn -Prelease-all -DskipTests -Dspotbugs.skiptrue clean install -U第5步骤正确后进入到 /rocketmq-all-5.1.3-source-release/distribution/target/rocketmq-5.1.3/rocketmq-5.1.3目录下然后启动NameServer执行如下命令 cd /root/rocketmq-all-5.1.3-source-release/distribution/target/rocketmq-5.1.3/rocketmq-5.1.3
nohup sh bin/mqnamesrv 验证NameServer是否启动成功执行如下命令 tail -f ~/logs/rocketmqlogs/namesrv.log会看到如下内容说明已经正常启动了 The Name Server boot success. serializeTypeJSON, address 0.0.0.0:9876 或者执行jps命令查看是否已经有了NameServer进程NamesrvStartup如有说明ok 第5、6步骤正确后进入到 /rocketmq-all-5.1.3-source-release/distribution/target/rocketmq-5.1.3/rocketmq-5.1.3目录下然后启动Broker和Proxy执行如下命令注意NameServer成功启动后我们启动Broker和Proxy5.x 版本下我们建议使用 Local 模式部署即 Broker 和 Proxy 同进程部署。5.x 版本也支持 Broker 和 Proxy 分离部署以实现更灵活的集群能力。详情参考其他教程。 cd /root/rocketmq-all-5.1.3-source-release/distribution/target/rocketmq-5.1.3/rocketmq-5.1.3
nohup sh bin/mqbroker -n localhost:9876 --enable-proxy 验证NameServer是否启动成功执行如下命令 tail -f ~/logs/rocketmqlogs/proxy.log会看到如下内容说明已经正常启动了 The broker[broker-a, 192.168.36.132:10911] boot success. serializeTypeJSON and name server is localhost:9876 或者执行jps命令查看是否已经有了ProxyStartup 进程如有说明ok
三、Java API 编写Producer和Consumer进行测试
上述正常启动NameServer和Broker及Proxy后首先需要创建名为TestTopic的Topic执行如下命令cd /root/rocketmq-all-5.1.3-source-release/distribution/target/rocketmq-5.1.3/rocketmq-5.1.3
sh bin/mqadmin updatetopic -n localhost:9876 -t TestTopic -c DefaultCluster查看新创建的Topic验证是否已经创建好执行sh bin/mqadmin topicList -n localhost:9876结果如下 创建消费者组执行如下命令cd /root/rocketmq-all-5.1.3-source-release/distribution/target/rocketmq-5.1.3/rocketmq-5.1.3
sh bin/sh mqadmin updateSubGroup -g testgroup -c DefaultCluster -n localhost:9876执行命令无任何错误即说明已经创建成功。在Idea中创建Maven工程添加rocketmq依赖添加如下依赖到pom.xml中propertiesmaven.compiler.source8/maven.compiler.sourcemaven.compiler.target8/maven.compiler.targetproject.build.sourceEncodingUTF-8/project.build.sourceEncodingrocketmq-client-java-version5.0.5/rocketmq-client-java-versionslf4j.version1.7.25/slf4j.version
/propertiesdependenciesdependencygroupIdorg.apache.rocketmq/groupIdartifactIdrocketmq-client-java/artifactIdversion${rocketmq-client-java-version}/version/dependencydependencygroupIdorg.slf4j/groupIdartifactIdslf4j-api/artifactIdversion${slf4j.version}/version/dependencydependencygroupIdorg.slf4j/groupIdartifactIdslf4j-log4j12/artifactIdversion${slf4j.version}/version/dependency
/dependencies编写ProducerTest生产者代码如下import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import java.io.IOException;public class ProducerTest {private static final Logger logger LoggerFactory.getLogger(ProducerTest.class);public static void main(String[] args) throws Exception {testMain();}public static void testMain() throws ClientException, IOException {// 接入点地址需要设置成Proxy的地址和端口列表一般是xxx:8081;xxx:8081。String endpoint 192.168.36.132:8081;// 消息发送的目标Topic名称需要提前创建。// 执行sh bin/mqadmin updatetopic -n localhost:9876 -t TestTopic -c DefaultClusterString topic TestTopic;ClientServiceProvider provider ClientServiceProvider.loadService();ClientConfigurationBuilder builder ClientConfiguration.newBuilder().setEndpoints(endpoint);ClientConfiguration configuration builder.build();// 初始化Producer时需要设置通信配置以及预绑定的Topic。Producer producer provider.newProducerBuilder().setTopics(topic).setClientConfiguration(configuration).build();int temp 0;while (true) {String msg 第 temp 条消息我喜欢rocketmq;temp;// 普通消息发送。Message message provider.newMessageBuilder().setTopic(topic)// 设置消息索引键可根据关键字精确查找某条消息。.setKeys(messageKey)// 设置消息Tag用于消费端根据指定Tag过滤消息。.setTag(messageTag)// 消息体。.setBody(msg.getBytes()).build();try {// 发送消息需要关注发送结果并捕获失败等异常。SendReceipt sendReceipt producer.send(message);Thread.sleep(1000);logger.info(Send message successfully, messageId{}, sendReceipt.getMessageId());} catch (Exception e) {logger.error(Failed to send message, e);}}// producer.close();}
}编写CommonUtils工具类用于将ByteBuffer转成String代码如下import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;public class CommonUtils {public static void main(String[] args) {System.out.println(Hello world!);}public static String decodeKey(ByteBuffer bytes) {Charset charset StandardCharsets.UTF_8;return charset.decode(bytes).toString();}public static byte[] decodeValue(ByteBuffer bytes) {int len bytes.limit() - bytes.position();byte[] bytes1 new byte[len];bytes.get(bytes1);return bytes1;}public static ByteBuffer encodeKey(String key) {return ByteBuffer.wrap(key.getBytes(StandardCharsets.UTF_8));}public static ByteBuffer encodeValue(byte[] value) {ByteBuffer byteBuffer ByteBuffer.allocate(value.length);byteBuffer.clear();byteBuffer.get(value, 0, value.length);return byteBuffer;}
}编写ConsumerTest生产者代码如下import java.util.Collections;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.rocketmq.producer.CommonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;public class PushConsumerTest {private static final Logger logger LoggerFactory.getLogger(PushConsumerTest.class);private PushConsumerTest() {}public static void main(String[] args) throws Exception {final ClientServiceProvider provider ClientServiceProvider.loadService();// 接入点地址需要设置成Proxy的地址和端口列表一般是xxx:8081;xxx:8081。String endpoints 192.168.36.132:8081;ClientConfiguration clientConfiguration ClientConfiguration.newBuilder().setEndpoints(endpoints).build();// 订阅消息的过滤规则表示订阅所有Tag的消息。String tag *;FilterExpression filterExpression new FilterExpression(tag, FilterExpressionType.TAG);// 为消费者指定所属的消费者分组Group需要提前创建。// 执行sh bin/sh mqadmin updateSubGroup -g testgroup -c DefaultCluster -n localhost:9876String consumerGroup testgroup;// 指定需要订阅哪个目标TopicTopic需要提前创建。String topic TestTopic;// 初始化PushConsumer需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。PushConsumer pushConsumer provider.newPushConsumerBuilder().setClientConfiguration(clientConfiguration)// 设置消费者分组。.setConsumerGroup(consumerGroup)// 设置预绑定的订阅关系。.setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))// 设置消费监听器。.setMessageListener(messageView - {// 处理消息并返回消费结果。logger.info(Consume message successfully, messageId{},messageBody{}, messageView.getMessageId(), CommonUtils.decodeKey(messageView.getBody()));return ConsumeResult.SUCCESS;}).build();Thread.sleep(Long.MAX_VALUE);// 如果不需要再使用 PushConsumer可关闭该实例。// pushConsumer.close();}
}为了能查看到控制台日志输入需要在resources目录下新建log4j.properties、log4j2.properties具体内容如下 log4j.properties内容log4j.rootLoggerINFO,consolelog4j.appender.consoleorg.apache.log4j.ConsoleAppender
log4j.appender.console.targetSystem.out
log4j.appender.console.layoutorg.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern%-4r [%t] %-5p %c %x - %m%nlog4j2.properties内容name PropertiesConfig
property.filename target/logs#appenders console, file
#配置值是appender的类型并不是具体appender实例的name
appenders rollingappender.rolling.type RollingFile
appender.rolling.name RollingLogFile
appender.rolling.fileName${filename}/automationlogs.log
appender.rolling.filePattern ${filename}/automationlogs-%d{MM-dd-yy-HH-mm-ss}-%i.log
appender.rolling.layout.type PatternLayout
appender.rolling.layout.pattern[%-5level] %d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %c{1} - %msg%n
appender.rolling.policies.type Policies
appender.rolling.policies.size.type SizeBasedTriggeringPolicy
appender.rolling.policies.size.size100MB
appender.rolling.strategy.type DefaultRolloverStrategy
appender.rolling.strategy.max 5rootLogger.level INFO,console
rootLogger.appenderRef.rolling.ref RollingLogFile到此完成了所有准备工作了整个工程如下所示 运行ProducerTest程序进行消息的发送控制台中会看到如下内容 运行ConsumerTest程序接收消息控制台中会看到如下内容
四、小结 至此一个单节点副本的 RocketMQ 集群已经部署起来了我们也通过编写Java程序进行简单的消息收发。如本文对您有帮助麻烦您动动发财的手指点个赞~~~~~谢谢您的阅读
