topic 配置对应实体
package com.alibaba.rocketmq.common;
import com.alibaba.rocketmq.common.constant.PermName;
public class TopicConfig {
public static int DefaultReadQueueNums = 16;
public static int DefaultWriteQueueNums = 16;
private static final String SEPARATOR = " ";
private String topicName;
private int readQueueNums;
private int writeQueueNums;
/**
* 6:同时支持读写
* 4:禁写
* 2:禁读
*/
private int perm; //读写权限
private TopicFilterType topicFilterType;
private int topicSysFlag;
private boolean order;
}
topic perm 默认为 6 可写可读,禁写就是要把broker中的topic 对应的 TopicConfig perm 设置为4 ,broker会上报到nameserver
api
查询和修改topic配置的rpc请求code为
RequestCode.UPDATE_AND_CREATE_TOPIC = 17 //更新TopicConfig
RequestCode.GET_ALL_TOPIC_CONFIG = 21 //拉取TopicConfig (全量)
调用
仅使用netty客户端即可
public class TopicConfig {
private final static NettyClientConfig clientConfig = new NettyClientConfig();
//netty客户端
private final static NettyRemotingClient remotingClient = new NettyRemotingClient(clientConfig);
public static void main(String[] args) throws InterruptedException, MQBrokerException, RemotingException, MQClientException {
remotingClient.start();
String topic = "quickStart"; //topic名称
String brokerAddr = "XX.XX.X.XXX:10910"; //broker地址
TopicConfigSerializeWrapper allTopicConfig = getAllTopicConfig(brokerAddr);
com.alibaba.rocketmq.common.TopicConfig config = allTopicConfig.getTopicConfigTable().get(topic);
config.setPerm(4); //禁写设为4 恢复设为6 禁读设为2
createAndUpdateTopic(brokerAddr, MixAll.DEFAULT_TOPIC, config, clientConfig.getConnectTimeoutMillis());
System.out.println("done");
}
/**
* 更新或创建
* @param addr
* @param defaultTopic
* @param topicConfig
* @param timeoutMillis
* @throws RemotingException
* @throws MQBrokerException
* @throws InterruptedException
* @throws MQClientException
*/
public static void createAndUpdateTopic(final String addr, final String defaultTopic, final com.alibaba.rocketmq.common.TopicConfig topicConfig,
final long timeoutMillis) throws RemotingException, MQBrokerException, InterruptedException,
MQClientException {
CreateTopicRequestHeader requestHeader = new CreateTopicRequestHeader();
requestHeader.setTopic(topicConfig.getTopicName());
requestHeader.setDefaultTopic(defaultTopic);
requestHeader.setReadQueueNums(topicConfig.getReadQueueNums());
requestHeader.setWriteQueueNums(topicConfig.getWriteQueueNums());
requestHeader.setPerm(topicConfig.getPerm());
requestHeader.setTopicFilterType(topicConfig.getTopicFilterType().name());
requestHeader.setTopicSysFlag(topicConfig.getTopicSysFlag());
requestHeader.setOrder(topicConfig.isOrder());
RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.UPDATE_AND_CREATE_TOPIC, requestHeader);
RemotingCommand response = remotingClient.invokeSync(addr, request, timeoutMillis);
assert response != null;
switch (response.getCode()) {
case ResponseCode.SUCCESS: {
return;
}
default:
break;
}
throw new MQClientException(response.getCode(), response.getRemark());
}
/**
* 拉取
* @param addr
* @return
* @throws RemotingConnectException
* @throws RemotingSendRequestException
* @throws RemotingTimeoutException
* @throws InterruptedException
* @throws MQBrokerException
*/
public static TopicConfigSerializeWrapper getAllTopicConfig(String addr) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQBrokerException {
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, (CommandCustomHeader) null);
RemotingCommand response = remotingClient.invokeSync(addr, request, 3000L);
assert response != null;
switch (response.getCode()) {
case 0:
return TopicConfigSerializeWrapper.decode(response.getBody(), TopicConfigSerializeWrapper.class);
default:
throw new MQBrokerException(response.getCode(), response.getRemark());
}
}
}
需要用broker的类,TopicConfigSerializeWrapper
<dependency>
<groupId>com.alibaba.rocketmq</groupId>
<artifactId>rocketmq-broker</artifactId>
<version>${rocketmq.version}</version>
</dependency>
网友评论