添加rocketmq线程池支持

pull/246/head
章新好 Sunny Zhang 2 years ago
parent 78dfd574d4
commit d7480f8105

@ -20,9 +20,21 @@ package cn.hippo4j.adapter.rocketmq;
import cn.hippo4j.adapter.base.ThreadPoolAdapter;
import cn.hippo4j.adapter.base.ThreadPoolAdapterParameter;
import cn.hippo4j.adapter.base.ThreadPoolAdapterState;
import cn.hippo4j.common.config.ApplicationContextHolder;
import cn.hippo4j.common.toolkit.ReflectUtil;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.impl.consumer.ConsumeMessageService;
import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer;
import org.springframework.boot.context.event.ApplicationStartedEvent;
import org.springframework.context.ApplicationListener;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ThreadPoolExecutor;
import static cn.hippo4j.common.constant.ChangeThreadPoolConstants.CHANGE_DELIMITER;
/**
* RocketMQ thread-pool adapter.
@ -30,6 +42,8 @@ import org.springframework.context.ApplicationListener;
@Slf4j
public class RocketMQThreadPoolAdapter implements ThreadPoolAdapter, ApplicationListener<ApplicationStartedEvent> {
private final Map<String, ThreadPoolExecutor> ROCKET_MQ_CONSUME_EXECUTOR = Maps.newHashMap();
@Override
public String mark() {
return "RocketMQ";
@ -37,16 +51,61 @@ public class RocketMQThreadPoolAdapter implements ThreadPoolAdapter, Application
@Override
public ThreadPoolAdapterState getThreadPoolState(String identify) {
return null;
ThreadPoolAdapterState result = new ThreadPoolAdapterState();
ThreadPoolExecutor rocketMQConsumeExecutor = ROCKET_MQ_CONSUME_EXECUTOR.get(identify);
if (rocketMQConsumeExecutor != null) {
result.setThreadPoolKey(identify);
result.setCoreSize(rocketMQConsumeExecutor.getCorePoolSize());
result.setMaximumSize(rocketMQConsumeExecutor.getMaximumPoolSize());
return result;
}
log.warn("[{}] RocketMQ consuming thread pool not found.", identify);
return result;
}
@Override
public List<ThreadPoolAdapterState> getThreadPoolStates() {
List<ThreadPoolAdapterState> adapterStateList = Lists.newArrayList();
ROCKET_MQ_CONSUME_EXECUTOR.forEach(
(key, val) -> adapterStateList.add(getThreadPoolState(key)));
return adapterStateList;
}
@Override
public boolean updateThreadPool(ThreadPoolAdapterParameter threadPoolAdapterParameter) {
String threadPoolKey = threadPoolAdapterParameter.getThreadPoolKey();
ThreadPoolExecutor rocketMQConsumeExecutor = ROCKET_MQ_CONSUME_EXECUTOR.get(threadPoolKey);
if (rocketMQConsumeExecutor != null) {
int originalCoreSize = rocketMQConsumeExecutor.getCorePoolSize();
int originalMaximumPoolSize = rocketMQConsumeExecutor.getMaximumPoolSize();
rocketMQConsumeExecutor.setCorePoolSize(threadPoolAdapterParameter.getCorePoolSize());
rocketMQConsumeExecutor.setMaximumPoolSize(threadPoolAdapterParameter.getMaximumPoolSize());
log.info("[{}] RocketMQ consumption thread pool parameter change. coreSize :: {}, maximumSize :: {}",
threadPoolKey,
String.format(CHANGE_DELIMITER, originalCoreSize, rocketMQConsumeExecutor.getCorePoolSize()),
String.format(CHANGE_DELIMITER, originalMaximumPoolSize, rocketMQConsumeExecutor.getMaximumPoolSize()));
return true;
}
log.warn("[{}] RocketMQ consuming thread pool not found.", threadPoolKey);
return false;
}
@Override
public void onApplicationEvent(ApplicationStartedEvent event) {
Map<String, DefaultRocketMQListenerContainer> containerMap =
ApplicationContextHolder.getBeansOfType(DefaultRocketMQListenerContainer.class);
try {
for (DefaultRocketMQListenerContainer container : containerMap.values()) {
DefaultMQPushConsumer consumer = container.getConsumer();
if (consumer != null) {
ConsumeMessageService consumeMessageService = consumer.getDefaultMQPushConsumerImpl().getConsumeMessageService();
ThreadPoolExecutor consumeExecutor = (ThreadPoolExecutor) ReflectUtil.getFieldValue(consumeMessageService, "consumeExecutor");
ROCKET_MQ_CONSUME_EXECUTOR.put(container.getConsumerGroup(), consumeExecutor);
}
}
} catch (Exception ex) {
log.error("Failed to get RocketMQ thread pool.", ex);
}
}
}

@ -0,0 +1,57 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>cn.hippo4j</groupId>
<artifactId>hippo4j-example</artifactId>
<version>${revision}</version>
</parent>
<artifactId>hippo4j-spring-boot-starter-adapter-rocketmq-example</artifactId>
<properties>
<maven.deploy.skip>true</maven.deploy.skip>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-json</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<dependency>
<groupId>cn.hippo4j</groupId>
<artifactId>hippo4j-spring-boot-starter-adapter-rocketmq</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>cn.hippo4j</groupId>
<artifactId>hippo4j-spring-boot-starter</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>cn.hippo4j</groupId>
<artifactId>hippo4j-example-core</artifactId>
<version>${revision}</version>
</dependency>
</dependencies>
</project>

@ -0,0 +1,31 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 cn.hippo4j.springboot.starter.adapter.rocketmq.example;
import cn.hippo4j.core.enable.EnableDynamicThreadPool;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@EnableDynamicThreadPool
@SpringBootApplication(scanBasePackages = {"cn.hippo4j.example.core", "cn.hippo4j.springboot.starter.adapter.rocketmq.example"})
public class Hippo4jAdapterRocketMQExampleApplication {
public static void main(String[] args) {
SpringApplication.run(Hippo4jAdapterRocketMQExampleApplication.class, args);
}
}

@ -0,0 +1,37 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 cn.hippo4j.springboot.starter.adapter.rocketmq.example;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;
/**
* Message consume.
*/
@Slf4j
@Service
@RocketMQMessageListener(topic = "test-topic", consumerGroup = "my-consumer_test-topic", consumeThreadNumber = 2)
public class MessageConsume implements RocketMQListener<String> {
@Override
public void onMessage(String message) {
log.info("Message: {}", message);
}
}

@ -0,0 +1,44 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 cn.hippo4j.springboot.starter.adapter.rocketmq.example;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.Date;
/**
* Message produce.
*/
@Slf4j
@Component
@RestController
@AllArgsConstructor
public class MessageProduce {
private final RocketMQTemplate rocketMQTemplate;
@GetMapping("/message/send")
public String sendMessage() {
rocketMQTemplate.convertAndSend("test-topic", new Date().toString());
return "success";
}
}

@ -0,0 +1,11 @@
server.port=8099
spring.profiles.active=dev
spring.dynamic.thread-pool.server-addr=http://localhost:6691
spring.dynamic.thread-pool.namespace=prescription
spring.dynamic.thread-pool.item-id=dynamic-threadpool-example
spring.dynamic.thread-pool.username=admin
spring.dynamic.thread-pool.password=123456
rocketmq.nameServer=127.0.0.1:9876
rocketmq.producer.group=my-producer

@ -22,6 +22,7 @@
<module>hippo4j-core-zookeeper-spring-boot-starter-example</module>
<module>hippo4j-spring-boot-starter-adapter-rabbitmq-example</module>
<module>hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rocketmq-example</module>
<module>hippo4j-spring-boot-starter-adapter-rocketmq-example</module>
</modules>
<properties>

Loading…
Cancel
Save