Rocketmq在Java代码之中手动创建Topic
【原创，转载请注明出处】
我的 【博客园主页】 【CSDN主页】 【简书主页】
加V进Java交流群，备注Java交流：w1129574379
** 本文仅限RocketMQ 4.5.*版本，其他版本可能有区别，仅供参考 **
本文仅限单 name server 的情况，nameserver集群的情况下不确定能否正常工作(原理都是一样的)
参考信息源及相关类/方法
org.apache.rocketmq.tools.command.topic.UpdateTopicSubCommand#execute
org.apache.rocketmq.client.ClientConfig
额外的依赖
需要添加额外的maven依赖，版本自选
org.apache.rocketmq
rocketmq-tools
4.5.1
org.apache.rocketmq
rocketmq-spring-boot-starter
2.0.3
使用
在命令行中创建Topic时应使用updateTopic命令，其使用示例如下：
./mqadmin updateTopic -n localhost:9876 -b localhost:10911 -t tx-mq-TOPIC
各参数含义如下：
mqadmin updateTopic [-b ] [-c ] [-h] [-n ] [-o ] [-p ] [-r ] [-s ]
-t [-u ] [-w ]
-b,--brokerAddr create topic to which broker
-c,--clusterName create topic to which cluster
-h,--help Print help
-n,--namesrvAddr Name server address list, eg: 192.168.0.1:9876;192.168.0.2:9876
-o,--order set topic's order(true|false)
-p,--perm set topic's permission(2|4|6), intro[2:W 4:R; 6:RW]
-r,--readQueueNums set read queue nums
-s,--hasUnitSub has unit sub (true|false)
-t,--topic topic name
-u,--unit is unit topic (true|false)
-w,--writeQueueNums set write queue nums
注意事项及存在的问题：
RocketMQ规定，在使用updateTopic命令创建topic时，-b或-c选项必须指定其中一个(都指定则处理-b参数，忽略-c参数)，与此同时，-t参数也为必要参数，缺少这几个必要参数则topic创建失败！！！
使用Java代码手动创建topic的形式中，使用-b选项创建topic可以正常使用，此时RocketMQ直接使用指定的broker地址来找到broker并在对应broker上创建topic。
使用Java代码手动创建topic的形式中，直接使用-c选项创建topic无法创建成功，因为org.apache.rocketmq.tools.command.topic.UpdateTopicSubCommand#execute方法需要一个DefaultMQAdminExt对象来连接到对应nameserver上以便获取对应集群下的所有broker信息，
DefaultMQAdminExt对象针对nameserver的处理代码为：private String namesrvAddr = NameServerAddressUtils.getNameServerAddresses()，查看代码发现其实际实现为System.getProperty(MixAll.NAMESRV_ADDR_PROPERTY, System.getenv(MixAll.NAMESRV_ADDR_ENV))。此时问题出现了，这里的实现代码并不能get到namesrv地址的参数值，导致连接到namesrv失败，完整报错信息如下：
点击查看完整报错信息
org.apache.rocketmq.tools.command.SubCommandException: UpdateTopicSubCommand command failed
at org.apache.rocketmq.tools.command.topic.UpdateTopicSubCommand.execute(UpdateTopicSubCommand.java:185)
at com.bayss.bws.common.utils.RocketMQUtil.createTopic(RocketMQUtil.java:54)
at com.bayss.bws.agent.internal.ConsumerManager.init(ConsumerManager.java:78)
at com.bayss.bws.agent.core.InitializeAgent.init(InitializeAgent.java:74)
at com.bayss.bws.agent.core.InitializeAgent.onApplicationEvent(InitializeAgent.java:184)
at org.springframework.context.event.SimpleApplicationEventMulticaster.doInvokeListener(SimpleApplicationEventMulticaster.java:172)
at org.springframework.context.event.SimpleApplicationEventMulticaster.invokeListener(SimpleApplicationEventMulticaster.java:165)
at org.springframework.context.event.SimpleApplicationEventMulticaster.multicastEvent(SimpleApplicationEventMulticaster.java:139)
at org.springframework.context.support.AbstractApplicationContext.publishEvent(AbstractApplicationContext.java:402)
at org.springframework.context.support.AbstractApplicationContext.publishEvent(AbstractApplicationContext.java:359)
at org.springframework.context.support.AbstractApplicationContext.finishRefresh(AbstractApplicationContext.java:896)
at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:552)
at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:775)
at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:397)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:316)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1260)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1248)
at com.bayss.bws.agent.core.BwsAgentApplication.main(BwsAgentApplication.java:24)
Caused by: org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to failed
at org.apache.rocketmq.remoting.netty.NettyRemotingClient.invokeSync(NettyRemotingClient.java:392)
at org.apache.rocketmq.client.impl.MQClientAPIImpl.getBrokerClusterInfo(MQClientAPIImpl.java:1193)
at org.apache.rocketmq.tools.admin.DefaultMQAdminExtImpl.examineBrokerClusterInfo(DefaultMQAdminExtImpl.java:275)
at org.apache.rocketmq.tools.admin.DefaultMQAdminExt.examineBrokerClusterInfo(DefaultMQAdminExt.java:222)
at org.apache.rocketmq.tools.command.CommandUtil.fetchMasterAddrByClusterName(CommandUtil.java:83)
at org.apache.rocketmq.tools.command.topic.UpdateTopicSubCommand.execute(UpdateTopicSubCommand.java:158)
... 17 more
于是决定采用手动set属性值的方式(对应属性值设值完成后不再需要-n参数即可创建topic)，手动set属性值的代码如下:
@Configuration
public class RocketMQConfig implements InitializingBean {
// 必须保证这里能获取到正确的namesrv地址，否则再次gg
@Value("${rocketmq.name-server}")
private String rocketMQNamesrv;
@Override
public void afterPropertiesSet() throws Exception {
System.setProperty(MixAll.NAMESRV_ADDR_PROPERTY, rocketMQNamesrv);
}
}
在命令行中使用updateTopic命令帮助时，有-n选项，但是在RocketMQ源码中并没有发现该选项的处理逻辑，并且在只设置-n(namesrv)时程序会报错(因为缺少第一条所说的-b或-c参数)。
使用示例：
String[] subargs = new String[] {
"-b 10.1.4.231:10911",
"-t unit-test-from-java-111",
//"-r 8",
//"-w 8",
//"-p 6",
//"-o false",
//"-u false",
//"-s false"
};
boolean isTopicCreated = RocketMQUtil.createTopic(subargs);
//boolean isTopicCreated = RocketMQUtil.createTopic("10.1.4.231:10911", "", "testttttt");
if (isTopicCreated) {
System.err.println("topic create success");
}
Java工具类
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.PosixParser;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.remoting.exception.RemotingConnectException;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.srvutil.ServerUtil;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.apache.rocketmq.tools.command.SubCommandException;
import org.apache.rocketmq.tools.command.topic.UpdateTopicSubCommand;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
public class RocketMQUtil {
public static boolean createTopic(String[] subargs) throws SubCommandException {
UpdateTopicSubCommand cmd = new UpdateTopicSubCommand();
Options options = ServerUtil.buildCommandlineOptions(new Options());
final Options updateTopicOptions = cmd.buildCommandlineOptions(options);
final CommandLine commandLine = ServerUtil
.parseCmdLine("mqadmin " cmd.commandName(),
subargs, updateTopicOptions, new PosixParser());
cmd.execute(commandLine, updateTopicOptions, null);
return true;
}
public static boolean createTopic(String brokerAddr, String clusterName, String topic) throws Exception {
if (StringUtils.isBlank(topic)) {
return false;
}
List argList = new LinkedList<>();
argList.add("-t " topic);
if (StringUtils.isNotBlank(brokerAddr)) {
argList.add("-b " brokerAddr.trim());
} else {
argList.add("-c " clusterName.trim());
}
return createTopic(argList.toArray(new String [0]));
}
public static boolean createTopic(String namesrvAddr, String topic) {
try {
Set clusterNames = RocketMQUtil.getClusterNames(namesrvAddr);
for (String clusterName : clusterNames) {
RocketMQUtil.createTopic(null, clusterName, topic);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
public static ClusterInfo getClusterInfo(String namesrvAddr) throws MQClientException, InterruptedException, MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
if (StringUtils.isBlank(namesrvAddr)) {
return new ClusterInfo();
}
DefaultMQAdminExt mqAdminExt = new DefaultMQAdminExt(5000L);
mqAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
mqAdminExt.setNamesrvAddr(namesrvAddr);
mqAdminExt.start();
ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo();
mqAdminExt.shutdown();
return clusterInfo;
}
public static Set getClusterNames(String namesrvAddr) throws MQClientException, InterruptedException, MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
return getClusterInfo(namesrvAddr).getClusterAddrTable().keySet();
}
public static Map getAllBrokerInfo(String namesrvAddr) throws MQClientException, InterruptedException, MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
return getClusterInfo(namesrvAddr).getBrokerAddrTable();
}
public static Set getBrokerAddrs(String namesrvAddr) throws InterruptedException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQClientException, MQBrokerException {
Map allBrokerInfo = getAllBrokerInfo(namesrvAddr);
Set brokerAddrs = new HashSet<>();
for (BrokerData brokerData : allBrokerInfo.values()) {
brokerAddrs.addAll(brokerData.getBrokerAddrs().values());
}
return brokerAddrs;
}
}