RocketMQ V5.x Java SDK收发消息报CancellationException问题求助

问题场景 #

已按官方《快速开始》配置RocketMQ，确认8081端口的gRPC服务正常运行，但使用Java SDK发送消息时抛出异常：

java.lang.IllegalStateException: Expected the service ProducerImpl-0 [FAILED] to be RUNNING, but the service has FAILED
    at org.apache.rocketmq.shaded.com.google.common.util.concurrent.AbstractService.checkCurrentState(AbstractService.java:381)
    ...
Caused by: java.util.concurrent.CancellationException: Task was cancelled.
    at org.apache.rocketmq.shaded.com.google.common.util.concurrent.AbstractFuture.cancellationExceptionWithCause(AbstractFuture.java:1543)
    ...

发送消息核心代码：

public static void sendMessage(String topic, String message) {
    try {
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfiguration configuration = ClientConfiguration.newBuilder()
                .setEndpoints("ip:8081")
                .build();
        Producer producer = provider.newProducerBuilder()
                .setClientConfiguration(configuration)
                .setTopics(topic)
                .build();
        String tag = "messageTag";
        Message msg = provider.newMessageBuilder()
                .setTopic(topic)
                .setTag(tag)
                .setBody(message.getBytes(StandardCharsets.UTF_8))
                .build();
        SendReceipt send = producer.send(msg);
        log.info("Send message successfully, messageId={}", send.getMessageId());
        producer.close();
    } catch (Exception e) {
        log.error(e.getMessage(), e);
        throw new RuntimeException(e.getMessage());
    }
}

排查与解决方案 #

1. 修正端点协议前缀 #

RocketMQ 5.x的gRPC端点需要明确指定协议前缀，修改ClientConfiguration的端点配置：

ClientConfiguration configuration = ClientConfiguration.newBuilder()
        .setEndpoints("http://ip:8081")
        .build();

若启用TLS加密，替换为https://ip:8081。

2. 确认Topic已创建 #

确保代码中指定的topic已在RocketMQ控制台或通过命令行创建完成，Producer启动时会校验Topic存在性，未创建会导致初始化失败。

3. 对齐SDK与服务端版本 #

Java SDK版本必须与RocketMQ服务端版本完全一致，版本不匹配会引发初始化异常。例如服务端为5.1.4，SDK需使用org.apache.rocketmq:rocketmq-client-java:5.1.4。

4. 添加Producer启动等待逻辑 #

Producer构建后需要时间完成初始化，直接调用send可能导致服务未就绪，可添加状态等待：

Producer producer = provider.newProducerBuilder()
        .setClientConfiguration(configuration)
        .setTopics(topic)
        .build();
// 等待Producer进入RUNNING状态，超时时间可按需调整
if (!producer.waitForRunning(Duration.ofSeconds(30))) {
    throw new IllegalStateException("Producer启动超时");
}

5. 排查网络连通性 #

验证应用服务器与RocketMQ服务端的8081端口通信正常，可通过telnet ip 8081或curl http://ip:8081/health测试，确保防火墙/安全组放开该端口双向访问权限。

6. 查看服务端日志 #

登录RocketMQ服务端，检查rocketmq/logs/proxy.log，搜索客户端连接相关错误（如认证失败、权限不足），这类问题会直接导致Producer初始化失败。

内容的提问来源于stack exchange，提问作者EasonHu