GitHub user 1270045840 added a comment to the discussion: 启动报错Expected the
service PushConsumerImpl-0 [FAILED] to be RUNNING, but the service has FAILED
public class PushConsumerExample {
private static final Logger logger =
LoggerFactory.getLogger(PushConsumerExample.class);
private PushConsumerExample() {
}
public static void main(String[] args) throws ClientException, IOException,
InterruptedException {
final ClientServiceProvider provider =
ClientServiceProvider.loadService();
// 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8081;xxx:8081。
String endpoints = "127.0.0.1:8081";
ClientConfiguration clientConfiguration =
ClientConfiguration.newBuilder()
.setEndpoints(endpoints)
.build();
// 订阅消息的过滤规则，表示订阅所有Tag的消息。
String tag = "*";
FilterExpression filterExpression = new FilterExpression(tag,
FilterExpressionType.TAG);
// 为消费者指定所属的消费者分组，Group需要提前创建。
String consumerGroup = "userLog";
// 指定需要订阅哪个目标Topic，Topic需要提前创建。
String topic = "userLog";
// 初始化PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
PushConsumer pushConsumer = provider.newPushConsumerBuilder()
.setClientConfiguration(clientConfiguration)
// 设置消费者分组。
.setConsumerGroup(consumerGroup)
// 设置预绑定的订阅关系。
.setSubscriptionExpressions(Collections.singletonMap(topic,
filterExpression))
// 设置消费监听器。
.setMessageListener(messageView -> {
// 处理消息并返回消费结果。
logger.info("Consume message successfully, messageId={}",
messageView.getMessageId());
return ConsumeResult.SUCCESS;
})
.build();
// Thread.sleep(Long.MAX_VALUE);
// 如果不需要再使用 PushConsumer，可关闭该实例。
// pushConsumer.close();
}
}
您好，我切换为8081之后，还是一样的错误
GitHub link:
https://github.com/apache/rocketmq/discussions/6238#discussioncomment-5189544
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
