|
1 | | -/* |
2 | | - * Copyright 2012-2019 the original author or authors. |
3 | | - * |
4 | | - * Licensed under the Apache License, Version 2.0 (the "License"); |
5 | | - * you may not use this file except in compliance with the License. |
6 | | - * You may obtain a copy of the License at |
7 | | - * |
8 | | - * https://www.apache.org/licenses/LICENSE-2.0 |
9 | | - * |
10 | | - * Unless required by applicable law or agreed to in writing, software |
11 | | - * distributed under the License is distributed on an "AS IS" BASIS, |
12 | | - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
13 | | - * See the License for the specific language governing permissions and |
14 | | - * limitations under the License. |
15 | | - */ |
16 | | - |
17 | | -package org.ylzl.eden.dynamic.tp.spring.boot.adapter; |
18 | | - |
19 | | -import com.dtp.adapter.rocketmq.RocketMqDtpAdapter; |
20 | | -import com.dtp.common.ApplicationContextHolder; |
21 | | -import com.dtp.common.dto.ExecutorWrapper; |
22 | | -import com.dtp.common.util.ReflectionUtil; |
23 | | -import lombok.extern.slf4j.Slf4j; |
24 | | -import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; |
25 | | -import org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService; |
26 | | -import org.apache.rocketmq.client.impl.consumer.ConsumeMessageOrderlyService; |
27 | | -import org.apache.rocketmq.client.impl.consumer.ConsumeMessageService; |
28 | | -import org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl; |
29 | | -import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer; |
30 | | -import org.ylzl.eden.common.mq.integration.rocketmq.RocketMQConsumer; |
31 | | -import org.ylzl.eden.commons.collections.CollectionUtils; |
32 | | - |
33 | | -import java.util.Map; |
34 | | -import java.util.Objects; |
35 | | -import java.util.concurrent.ThreadPoolExecutor; |
36 | | - |
37 | | -/** |
38 | | - * 自定义 RocketMqDtpAdapter |
39 | | - * |
40 | | - * @author <a href="mailto:shiyindaxiaojie@gmail.com">gyl</a> |
41 | | - * @since 2.4.x |
42 | | - */ |
43 | | -@Slf4j |
44 | | -public class CustomRocketMqDtpAdapter extends RocketMqDtpAdapter { |
45 | | - |
46 | | - @Override |
47 | | - protected void initialize() { |
48 | | - Map<String, DefaultRocketMQListenerContainer> beans = ApplicationContextHolder.getBeansOfType(DefaultRocketMQListenerContainer.class); |
49 | | - if (CollectionUtils.isNotEmpty(beans)) { |
50 | | - beans.forEach((k, v) -> { |
51 | | - DefaultMQPushConsumer consumer = v.getConsumer(); |
52 | | - String group = v.getConsumerGroup(); |
53 | | - String topic = v.getTopic(); |
54 | | - this.setThreadPoolExecutor(consumer, group, topic); |
55 | | - }); |
56 | | - } |
57 | | - |
58 | | - // support dynamic-mq |
59 | | - RocketMQConsumer rocketMQConsumer = ApplicationContextHolder.getBean(RocketMQConsumer.class); |
60 | | - if (rocketMQConsumer != null && !rocketMQConsumer.getConsumers().isEmpty()) { |
61 | | - rocketMQConsumer.getConsumers().forEach((k, v) -> { |
62 | | - this.setThreadPoolExecutor(v, v.getConsumerGroup(), k); |
63 | | - }); |
64 | | - } |
65 | | - |
66 | | - log.info("DynamicTp adapter, rocketMq consumer executors init end, executors: {}", this.executors); |
67 | | - } |
68 | | - |
69 | | - private void setThreadPoolExecutor(DefaultMQPushConsumer consumer, String group, String topic) { |
70 | | - DefaultMQPushConsumerImpl pushConsumer = consumer.getDefaultMQPushConsumerImpl(); |
71 | | -// DefaultMQPushConsumerImpl pushConsumer = (DefaultMQPushConsumerImpl) ReflectionUtil.getFieldValue(DefaultMQPushConsumer.class, "defaultMQPushConsumerImpl", consumer); |
72 | | - if (!Objects.isNull(pushConsumer)) { |
73 | | - String key = group + "#" + topic; |
74 | | - ThreadPoolExecutor executor = null; |
75 | | - ConsumeMessageService consumeMessageService = pushConsumer.getConsumeMessageService(); |
76 | | - if (consumeMessageService instanceof ConsumeMessageConcurrentlyService) { |
77 | | - executor = (ThreadPoolExecutor) ReflectionUtil.getFieldValue(ConsumeMessageConcurrentlyService.class, "consumeExecutor", consumeMessageService); |
78 | | - } else if (consumeMessageService instanceof ConsumeMessageOrderlyService) { |
79 | | - executor = (ThreadPoolExecutor) ReflectionUtil.getFieldValue(ConsumeMessageOrderlyService.class, "consumeExecutor", consumeMessageService); |
80 | | - } |
81 | | - |
82 | | - if (Objects.nonNull(executor)) { |
83 | | - ExecutorWrapper executorWrapper = new ExecutorWrapper(key, executor); |
84 | | - this.initNotifyItems(key, executorWrapper); |
85 | | - this.executors.put(key, executorWrapper); |
86 | | - } |
87 | | - } |
88 | | - } |
89 | | -} |
| 1 | +///* |
| 2 | +// * Copyright 2012-2019 the original author or authors. |
| 3 | +// * |
| 4 | +// * Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | +// * you may not use this file except in compliance with the License. |
| 6 | +// * You may obtain a copy of the License at |
| 7 | +// * |
| 8 | +// * https://www.apache.org/licenses/LICENSE-2.0 |
| 9 | +// * |
| 10 | +// * Unless required by applicable law or agreed to in writing, software |
| 11 | +// * distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | +// * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | +// * See the License for the specific language governing permissions and |
| 14 | +// * limitations under the License. |
| 15 | +// */ |
| 16 | +// |
| 17 | +//package org.ylzl.eden.dynamic.tp.spring.boot.adapter; |
| 18 | +// |
| 19 | +//import com.dtp.adapter.rocketmq.RocketMqDtpAdapter; |
| 20 | +//import com.dtp.common.ApplicationContextHolder; |
| 21 | +//import com.dtp.common.dto.ExecutorWrapper; |
| 22 | +//import com.dtp.common.util.ReflectionUtil; |
| 23 | +//import lombok.extern.slf4j.Slf4j; |
| 24 | +//import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; |
| 25 | +//import org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService; |
| 26 | +//import org.apache.rocketmq.client.impl.consumer.ConsumeMessageOrderlyService; |
| 27 | +//import org.apache.rocketmq.client.impl.consumer.ConsumeMessageService; |
| 28 | +//import org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl; |
| 29 | +//import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer; |
| 30 | +//import org.ylzl.eden.common.mq.integration.rocketmq.RocketMQConsumer; |
| 31 | +//import org.ylzl.eden.commons.collections.CollectionUtils; |
| 32 | +// |
| 33 | +//import java.util.Map; |
| 34 | +//import java.util.Objects; |
| 35 | +//import java.util.concurrent.ThreadPoolExecutor; |
| 36 | +// |
| 37 | +///** |
| 38 | +// * 自定义 RocketMqDtpAdapter |
| 39 | +// * |
| 40 | +// * @author <a href="mailto:shiyindaxiaojie@gmail.com">gyl</a> |
| 41 | +// * @since 2.4.x |
| 42 | +// */ |
| 43 | +//@Slf4j |
| 44 | +//public class CustomRocketMqDtpAdapter extends RocketMqDtpAdapter { |
| 45 | +// |
| 46 | +// @Override |
| 47 | +// protected void initialize() { |
| 48 | +// Map<String, DefaultRocketMQListenerContainer> beans = ApplicationContextHolder.getBeansOfType(DefaultRocketMQListenerContainer.class); |
| 49 | +// if (CollectionUtils.isNotEmpty(beans)) { |
| 50 | +// beans.forEach((k, v) -> { |
| 51 | +// DefaultMQPushConsumer consumer = v.getConsumer(); |
| 52 | +// String group = v.getConsumerGroup(); |
| 53 | +// String topic = v.getTopic(); |
| 54 | +// this.setThreadPoolExecutor(consumer, group, topic); |
| 55 | +// }); |
| 56 | +// } |
| 57 | +// |
| 58 | +// // support dynamic-mq |
| 59 | +// RocketMQConsumer rocketMQConsumer = ApplicationContextHolder.getBean(RocketMQConsumer.class); |
| 60 | +// if (rocketMQConsumer != null && !rocketMQConsumer.getConsumers().isEmpty()) { |
| 61 | +// rocketMQConsumer.getConsumers().forEach((k, v) -> { |
| 62 | +// this.setThreadPoolExecutor(v, v.getConsumerGroup(), k); |
| 63 | +// }); |
| 64 | +// } |
| 65 | +// |
| 66 | +// log.info("DynamicTp adapter, rocketMq consumer executors init end, executors: {}", this.executors); |
| 67 | +// } |
| 68 | +// |
| 69 | +// private void setThreadPoolExecutor(DefaultMQPushConsumer consumer, String group, String topic) { |
| 70 | +// DefaultMQPushConsumerImpl pushConsumer = consumer.getDefaultMQPushConsumerImpl(); |
| 71 | +//// DefaultMQPushConsumerImpl pushConsumer = (DefaultMQPushConsumerImpl) ReflectionUtil.getFieldValue(DefaultMQPushConsumer.class, "defaultMQPushConsumerImpl", consumer); |
| 72 | +// if (!Objects.isNull(pushConsumer)) { |
| 73 | +// String key = group + "#" + topic; |
| 74 | +// ThreadPoolExecutor executor = null; |
| 75 | +// ConsumeMessageService consumeMessageService = pushConsumer.getConsumeMessageService(); |
| 76 | +// if (consumeMessageService instanceof ConsumeMessageConcurrentlyService) { |
| 77 | +// executor = (ThreadPoolExecutor) ReflectionUtil.getFieldValue(ConsumeMessageConcurrentlyService.class, "consumeExecutor", consumeMessageService); |
| 78 | +// } else if (consumeMessageService instanceof ConsumeMessageOrderlyService) { |
| 79 | +// executor = (ThreadPoolExecutor) ReflectionUtil.getFieldValue(ConsumeMessageOrderlyService.class, "consumeExecutor", consumeMessageService); |
| 80 | +// } |
| 81 | +// |
| 82 | +// if (Objects.nonNull(executor)) { |
| 83 | +// ExecutorWrapper executorWrapper = new ExecutorWrapper(key, executor); |
| 84 | +// this.initNotifyItems(key, executorWrapper); |
| 85 | +// this.executors.put(key, executorWrapper); |
| 86 | +// } |
| 87 | +// } |
| 88 | +// } |
| 89 | +//} |
0 commit comments