ZhouXY108/hutool
1
0
Fork 0
mirror of https://gitee.com/chinabugotech/hutool.git synced 2025-04-19 03:01:48 +08:00
Code Issues Packages Projects Releases Wiki Activity

add rocketmq

Browse Source
This commit is contained in:
Looly 2025-01-28 18:09:59 +08:00
parent 253960ca73
commit 73659a0894
6 changed files with 277 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
hutool-extra/pom.xml
View File

@ -570,5 +570,11 @@
<version>5.24.0</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>5.3.1</version>
<scope>provided</scope>
</dependency>
</dependencies>
</project>

10
hutool-extra/src/main/java/org/dromara/hutool/extra/mq/engine/kafka/KafkaConsumer.java
View File

@ -90,10 +90,20 @@ public class KafkaConsumer implements Consumer {
IoUtil.nullSafeClose(this.consumer);
}
/**
* 消费者记录包装为消息
*
* @author looly
*/
private static class ConsumerRecordMessage implements Message {
private final ConsumerRecord<String, byte[]> record;
/**
* 构造
*
* @param record {@link ConsumerRecord}
*/
private ConsumerRecordMessage(final ConsumerRecord<String, byte[]> record) {
this.record = record;
}

106
hutool-extra/src/main/java/org/dromara/hutool/extra/mq/engine/rocketmq/RocketMQConsumer.java Normal file
View File

@ -0,0 +1,106 @@
/*
* Copyright (c) 2025 Hutool Team and hutool.cn
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.dromara.hutool.extra.mq.engine.rocketmq;
import org.apache.rocketmq.client.consumer.MQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.dromara.hutool.extra.mq.Consumer;
import org.dromara.hutool.extra.mq.MQException;
import org.dromara.hutool.extra.mq.Message;
import org.dromara.hutool.extra.mq.MessageHandler;
import java.io.IOException;
/**
* RocketMQ 消费者
*
* @author Looly
* @since 6.0.0
*/
public class RocketMQConsumer implements Consumer {
private final MQPushConsumer consumer;
/**
* 构造
*
* @param consumer RocketMQ PushConsumer
*/
public RocketMQConsumer(final MQPushConsumer consumer) {
this.consumer = consumer;
}
/**
* 设置消费的Topic
*
* @param topic Topic
* @return this
*/
public RocketMQConsumer setTopic(final String topic) {
try {
this.consumer.subscribe(topic, "*");
} catch (final MQClientException e) {
throw new MQException(e);
}
return this;
}
@Override
public void subscribe(final MessageHandler messageHandler) {
this.consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
for (final MessageExt msg : msgs) {
messageHandler.handle(new RocketMQMessage(msg));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
}
@Override
public void close() throws IOException {
if (null != this.consumer) {
this.consumer.shutdown();
}
}
/**
* RocketMQ消息包装
*
* @author Looly
* @since 6.0.0
*/
private static class RocketMQMessage implements Message {
private final MessageExt messageExt;
private RocketMQMessage(final MessageExt messageExt) {
this.messageExt = messageExt;
}
@Override
public String topic() {
return messageExt.getTopic();
}
@Override
public byte[] content() {
return messageExt.getBody();
}
}
}

71
hutool-extra/src/main/java/org/dromara/hutool/extra/mq/engine/rocketmq/RocketMQEngine.java Normal file
View File

@ -0,0 +1,71 @@
/*
* Copyright (c) 2025 Hutool Team and hutool.cn
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.dromara.hutool.extra.mq.engine.rocketmq;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.dromara.hutool.core.lang.Assert;
import org.dromara.hutool.extra.mq.Consumer;
import org.dromara.hutool.extra.mq.MQConfig;
import org.dromara.hutool.extra.mq.MQException;
import org.dromara.hutool.extra.mq.Producer;
import org.dromara.hutool.extra.mq.engine.MQEngine;
/**
* RocketMQ引擎
*
* @author Looly
* @since 6.0.0
*/
public class RocketMQEngine implements MQEngine {
private MQConfig config;
/**
* 默认构造
*/
public RocketMQEngine() {
// SPI方式加载时检查库是否引入
Assert.notNull( org.apache.rocketmq.common.message.Message.class);
}
@Override
public RocketMQEngine init(final MQConfig config) {
this.config = config;
return this;
}
@Override
public Producer getProducer() {
final DefaultMQProducer defaultMQProducer = new DefaultMQProducer();
defaultMQProducer.setNamesrvAddr(config.getBrokerUrl());
try {
defaultMQProducer.start();
} catch (final MQClientException e) {
throw new MQException(e);
}
return new RocketMQProducer(defaultMQProducer);
}
@Override
public Consumer getConsumer() {
final DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer();
defaultMQPushConsumer.setNamesrvAddr(config.getBrokerUrl());
return new RocketMQConsumer(defaultMQPushConsumer);
}
}

62
hutool-extra/src/main/java/org/dromara/hutool/extra/mq/engine/rocketmq/RocketMQProducer.java Normal file
View File

@ -0,0 +1,62 @@
/*
* Copyright (c) 2025 Hutool Team and hutool.cn
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.dromara.hutool.extra.mq.engine.rocketmq;
import org.apache.rocketmq.client.producer.MQProducer;
import org.dromara.hutool.extra.mq.MQException;
import org.dromara.hutool.extra.mq.Message;
import org.dromara.hutool.extra.mq.Producer;
import java.io.IOException;
/**
* RocketMQ Producer
*
* @author Looly
* @since 6.0.0
*/
public class RocketMQProducer implements Producer {
private final MQProducer producer;
/**
* 构造
*
* @param producer RocketMQ Producer
*/
public RocketMQProducer(final MQProducer producer) {
this.producer = producer;
}
@Override
public void send(final Message message) {
final org.apache.rocketmq.common.message.Message rocketMessage =
new org.apache.rocketmq.common.message.Message(message.topic(), message.content());
try {
this.producer.send(rocketMessage);
} catch (final Exception e) {
throw new MQException(e);
}
}
@Override
public void close() throws IOException {
if (null != this.producer) {
this.producer.shutdown();
}
}
}

22
hutool-extra/src/main/java/org/dromara/hutool/extra/mq/engine/rocketmq/package-info.java Normal file
View File

@ -0,0 +1,22 @@
/*
* Copyright (c) 2025 Hutool Team and hutool.cn
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
/**
* RocketMQ引擎
*
* @author Looly
*/
package org.dromara.hutool.extra.mq.engine.rocketmq;