From d58a7119effe92d63a311848d50f4dab8a2d4622 Mon Sep 17 00:00:00 2001 From: lijianxin <1064730540@qq.com> Date: Thu, 25 Aug 2022 10:59:22 +0800 Subject: [PATCH] feat: add support spring-cloud-stream-rabbitmq --- docs/docs/community/developer.md | 9 +- .../pom.xml | 55 ++++++ ...gCloudStreamRabbitMQThreadPoolAdapter.java | 166 ++++++++++++++++++ hippo4j-adapter/pom.xml | 1 + .../pom.xml | 78 ++++++++ .../rabbitmq/example/MessageProduce.java | 70 ++++++++ ...rSpringCloudStreamRabbitMQApplication.java | 52 ++++++ .../src/main/resources/application.properties | 28 +++ hippo4j-example/pom.xml | 1 + .../pom.xml | 51 ++++++ ...treamRabbitMQAdapterAutoConfiguration.java | 50 ++++++ .../main/resources/META-INF/spring.factories | 1 + .../pom.xml | 1 + pom.xml | 1 + 14 files changed, 563 insertions(+), 1 deletion(-) create mode 100644 hippo4j-adapter/hippo4j-adapter-spring-cloud-stream-rabbitmq/pom.xml create mode 100644 hippo4j-adapter/hippo4j-adapter-spring-cloud-stream-rabbitmq/src/main/java/cn/hippo4j/adapter/springcloud/stream/rabbitmq/SpringCloudStreamRabbitMQThreadPoolAdapter.java create mode 100644 hippo4j-example/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq-example/pom.xml create mode 100644 hippo4j-example/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq-example/src/main/java/cn/hippo4j/springboot/starter/adapter/springcloud/stream/rabbitmq/example/MessageProduce.java create mode 100644 hippo4j-example/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq-example/src/main/java/cn/hippo4j/springboot/starter/adapter/springcloud/stream/rabbitmq/example/ServerAdapterSpringCloudStreamRabbitMQApplication.java create mode 100644 hippo4j-example/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq-example/src/main/resources/application.properties create mode 100644 hippo4j-spring-boot/hippo4j-spring-boot-starter-adapter/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq/pom.xml create mode 100644 hippo4j-spring-boot/hippo4j-spring-boot-starter-adapter/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq/src/main/java/cn/hippo4j/springboot/starter/adapter/springcloud/stream/rabbitmq/SpringCloudStreamRabbitMQAdapterAutoConfiguration.java create mode 100644 hippo4j-spring-boot/hippo4j-spring-boot-starter-adapter/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq/src/main/resources/META-INF/spring.factories diff --git a/docs/docs/community/developer.md b/docs/docs/community/developer.md index 827427bc..bf813e09 100644 --- a/docs/docs/community/developer.md +++ b/docs/docs/community/developer.md @@ -45,5 +45,12 @@ sidebar_position: 2 核心开发者 weihubeats@163.com - + + + 李剑鑫 + BigXin0109 + Only丶Big + 核心开发者 + 1064730540@qq.com + diff --git a/hippo4j-adapter/hippo4j-adapter-spring-cloud-stream-rabbitmq/pom.xml b/hippo4j-adapter/hippo4j-adapter-spring-cloud-stream-rabbitmq/pom.xml new file mode 100644 index 00000000..d700d8a6 --- /dev/null +++ b/hippo4j-adapter/hippo4j-adapter-spring-cloud-stream-rabbitmq/pom.xml @@ -0,0 +1,55 @@ + + + 4.0.0 + + cn.hippo4j + hippo4j-adapter + ${revision} + + hippo4j-adapter-spring-cloud-stream-rabbitmq + + + + cn.hippo4j + hippo4j-adapter-base + + + + org.springframework.cloud + spring-cloud-starter-stream-rabbit + ${spring-cloud-starter-stream-rabbitmq.version} + + + + + + + org.apache.maven.plugins + maven-jar-plugin + + + + ${project.artifactId} + ${project.version} + ${maven.build.timestamp} + chen.ma + + + + + + org.apache.maven.plugins + maven-javadoc-plugin + 2.10.3 + + + + jar + + + + + + + diff --git a/hippo4j-adapter/hippo4j-adapter-spring-cloud-stream-rabbitmq/src/main/java/cn/hippo4j/adapter/springcloud/stream/rabbitmq/SpringCloudStreamRabbitMQThreadPoolAdapter.java b/hippo4j-adapter/hippo4j-adapter-spring-cloud-stream-rabbitmq/src/main/java/cn/hippo4j/adapter/springcloud/stream/rabbitmq/SpringCloudStreamRabbitMQThreadPoolAdapter.java new file mode 100644 index 00000000..edabaedf --- /dev/null +++ b/hippo4j-adapter/hippo4j-adapter-spring-cloud-stream-rabbitmq/src/main/java/cn/hippo4j/adapter/springcloud/stream/rabbitmq/SpringCloudStreamRabbitMQThreadPoolAdapter.java @@ -0,0 +1,166 @@ +/* + * 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.adapter.springcloud.stream.rabbitmq; + +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.CollectionUtil; +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.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; +import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer; +import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +import org.springframework.boot.context.event.ApplicationStartedEvent; +import org.springframework.cloud.stream.binder.Binding; +import org.springframework.cloud.stream.binder.DefaultBinding; +import org.springframework.cloud.stream.binding.InputBindingLifecycle; +import org.springframework.context.ApplicationListener; +import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter; + +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +import static cn.hippo4j.common.constant.ChangeThreadPoolConstants.CHANGE_DELIMITER; + +/** + * Spring cloud stream rabbimq thread-pool adapter. + */ +@Slf4j +public class SpringCloudStreamRabbitMQThreadPoolAdapter implements ThreadPoolAdapter, ApplicationListener { + + private final Map ROCKET_MQ_SPRING_CLOUD_STREAM_CONSUME_EXECUTOR = Maps.newHashMap(); + + @Override + public String mark() { + return "rabbitMQSpringCloudStream"; + } + + @Override + public ThreadPoolAdapterState getThreadPoolState(String identify) { + ThreadPoolAdapterState result = new ThreadPoolAdapterState(); + AbstractMessageListenerContainer messageListenerContainer = ROCKET_MQ_SPRING_CLOUD_STREAM_CONSUME_EXECUTOR.get(identify); + if (messageListenerContainer != null) { + result.setThreadPoolKey(identify); + if (messageListenerContainer instanceof SimpleMessageListenerContainer) { + int concurrentConsumers = (int) ReflectUtil.getFieldValue(messageListenerContainer, "concurrentConsumers"); + result.setCoreSize(concurrentConsumers); + Object maxConcurrentConsumers = ReflectUtil.getFieldValue(messageListenerContainer, "maxConcurrentConsumers"); + if (maxConcurrentConsumers != null) { + result.setMaximumSize((Integer) maxConcurrentConsumers); + } else { + result.setMaximumSize(concurrentConsumers); + } + + } else if (messageListenerContainer instanceof DirectMessageListenerContainer) { + int consumersPerQueue = (int) ReflectUtil.getFieldValue(messageListenerContainer, "consumersPerQueue"); + result.setCoreSize(consumersPerQueue); + result.setMaximumSize(consumersPerQueue); + } + return result; + } + log.warn("[{}] rabbitMQ consuming thread pool not found.", identify); + return result; + } + + @Override + public List getThreadPoolStates() { + List adapterStateList = Lists.newArrayList(); + ROCKET_MQ_SPRING_CLOUD_STREAM_CONSUME_EXECUTOR.forEach( + (key, val) -> adapterStateList.add(getThreadPoolState(key))); + return adapterStateList; + } + + @Override + public boolean updateThreadPool(ThreadPoolAdapterParameter threadPoolAdapterParameter) { + String threadPoolKey = threadPoolAdapterParameter.getThreadPoolKey(); + AbstractMessageListenerContainer messageListenerContainer = ROCKET_MQ_SPRING_CLOUD_STREAM_CONSUME_EXECUTOR.get(threadPoolKey); + if (messageListenerContainer != null) { + synchronized (ROCKET_MQ_SPRING_CLOUD_STREAM_CONSUME_EXECUTOR) { + Integer corePoolSize = threadPoolAdapterParameter.getCorePoolSize(); + Integer maximumPoolSize = threadPoolAdapterParameter.getMaximumPoolSize(); + if (messageListenerContainer instanceof SimpleMessageListenerContainer) { + int originalCoreSize = (int) ReflectUtil.getFieldValue(messageListenerContainer, "concurrentConsumers"); + Object maxConcurrentConsumers = ReflectUtil.getFieldValue(messageListenerContainer, "maxConcurrentConsumers"); + int originalMaximumPoolSize; + if (maxConcurrentConsumers != null) { + originalMaximumPoolSize = (Integer) maxConcurrentConsumers; + } else { + originalMaximumPoolSize = originalCoreSize; + } + SimpleMessageListenerContainer simpleMessageListenerContainer = (SimpleMessageListenerContainer) messageListenerContainer; + if (originalCoreSize > maximumPoolSize) { + simpleMessageListenerContainer.setConcurrentConsumers(corePoolSize); + simpleMessageListenerContainer.setMaxConcurrentConsumers(maximumPoolSize); + } else { + simpleMessageListenerContainer.setMaxConcurrentConsumers(maximumPoolSize); + simpleMessageListenerContainer.setConcurrentConsumers(corePoolSize); + } + log.info("[{}] rabbitMQ consumption thread pool parameter change. coreSize: {}, maximumSize: {}", + threadPoolKey, + String.format(CHANGE_DELIMITER, originalCoreSize, corePoolSize), + String.format(CHANGE_DELIMITER, originalMaximumPoolSize, maximumPoolSize)); + } else if (messageListenerContainer instanceof DirectMessageListenerContainer) { + int originalCoreSize = (int) ReflectUtil.getFieldValue(messageListenerContainer, "consumersPerQueue"); + DirectMessageListenerContainer directMessageListenerContainer = (DirectMessageListenerContainer) messageListenerContainer; + directMessageListenerContainer.setConsumersPerQueue(maximumPoolSize); + log.info("[{}] rabbitMQ consumption thread pool parameter change. coreSize: {}", + threadPoolKey, + String.format(CHANGE_DELIMITER, originalCoreSize, corePoolSize)); + } else { + log.warn("[{}] rabbitMQ consuming thread pool not support. messageListenerContainer: {}", threadPoolKey, messageListenerContainer.getClass()); + return false; + } + } + + return true; + } + log.warn("[{}] rabbitMQ consuming thread pool not found.", threadPoolKey); + return false; + } + + @Override + public void onApplicationEvent(ApplicationStartedEvent event) { + InputBindingLifecycle bindingLifecycle = ApplicationContextHolder.getBean(InputBindingLifecycle.class); + Collection> inputBindings = Optional.ofNullable(ReflectUtil.getFieldValue(bindingLifecycle, "inputBindings")) + .map(each -> (Collection>) each).orElse(null); + + if (CollectionUtil.isEmpty(inputBindings)) { + log.info("InputBindings record not found."); + return; + } + try { + for (Binding each : inputBindings) { + String bindingName = each.getBindingName(); + DefaultBinding defaultBinding = (DefaultBinding) each; + Object lifecycle = ReflectUtil.getFieldValue(defaultBinding, "lifecycle"); + if (lifecycle instanceof AmqpInboundChannelAdapter) { + AbstractMessageListenerContainer rabbitMQListenerContainer = (AbstractMessageListenerContainer) ReflectUtil.getFieldValue(lifecycle, "messageListenerContainer"); + ROCKET_MQ_SPRING_CLOUD_STREAM_CONSUME_EXECUTOR.put(bindingName, rabbitMQListenerContainer); + } + } + } catch (Exception ex) { + log.error("Failed to get input-bindings thread pool.", ex); + } + } +} diff --git a/hippo4j-adapter/pom.xml b/hippo4j-adapter/pom.xml index 2da1966a..d1c0624a 100644 --- a/hippo4j-adapter/pom.xml +++ b/hippo4j-adapter/pom.xml @@ -20,6 +20,7 @@ hippo4j-adapter-hystrix hippo4j-adapter-spring-cloud-stream-rocketmq hippo4j-adapter-spring-cloud-stream-kafka + hippo4j-adapter-spring-cloud-stream-rabbitmq hippo4j-adapter-web diff --git a/hippo4j-example/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq-example/pom.xml b/hippo4j-example/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq-example/pom.xml new file mode 100644 index 00000000..da01bedc --- /dev/null +++ b/hippo4j-example/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq-example/pom.xml @@ -0,0 +1,78 @@ + + + 4.0.0 + + cn.hippo4j + hippo4j-example + ${revision} + + hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq-example + + + true + + + + + org.springframework.boot + spring-boot-starter + + + + org.springframework.boot + spring-boot-starter-web + + + + org.springframework.boot + spring-boot-starter-json + + + + com.alibaba + fastjson + 1.2.76 + + + + org.projectlombok + lombok + + + + cn.hippo4j + hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq + ${project.version} + + + + cn.hippo4j + hippo4j-spring-boot-starter + ${project.version} + + + + cn.hippo4j + hippo4j-example-core + ${revision} + + + + + ${project.artifactId} + + + org.springframework.boot + spring-boot-maven-plugin + + + + repackage + + + + + + + diff --git a/hippo4j-example/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq-example/src/main/java/cn/hippo4j/springboot/starter/adapter/springcloud/stream/rabbitmq/example/MessageProduce.java b/hippo4j-example/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq-example/src/main/java/cn/hippo4j/springboot/starter/adapter/springcloud/stream/rabbitmq/example/MessageProduce.java new file mode 100644 index 00000000..29fb37d7 --- /dev/null +++ b/hippo4j-example/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq-example/src/main/java/cn/hippo4j/springboot/starter/adapter/springcloud/stream/rabbitmq/example/MessageProduce.java @@ -0,0 +1,70 @@ +/* + * 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.springcloud.stream.rabbitmq.example; + +import cn.hippo4j.example.core.dto.SendMessageDTO; +import com.alibaba.fastjson.JSON; +import lombok.AllArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.cloud.stream.function.StreamBridge; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.bind.annotation.RestController; + +import java.util.UUID; + +/** + * Message produce. + */ +@Slf4j +@RestController +@AllArgsConstructor +public class MessageProduce { + + private final StreamBridge streamBridge; + + @GetMapping("/message/send") + public String sendMessage(@RequestParam(required = false) Integer maxSendSize) { + if (maxSendSize == null) { + maxSendSize = 10; + } + for (int i = 0; i < maxSendSize; i++) { + sendMessage0(); + } + return "success"; + } + + private void sendMessage0() { + String keys = UUID.randomUUID().toString(); + SendMessageDTO payload = SendMessageDTO.builder() + .receiver("156011xxx91") + .uid(keys) + .build(); + long startTime = System.currentTimeMillis(); + boolean sendResult = false; + try { + sendResult = streamBridge.send("demoOutput", payload); + } finally { + log.info("Send status: {}, Keys: {}, Execute time: {} ms, Message: {}", + sendResult, + keys, + System.currentTimeMillis() - startTime, + JSON.toJSONString(payload)); + } + } +} diff --git a/hippo4j-example/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq-example/src/main/java/cn/hippo4j/springboot/starter/adapter/springcloud/stream/rabbitmq/example/ServerAdapterSpringCloudStreamRabbitMQApplication.java b/hippo4j-example/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq-example/src/main/java/cn/hippo4j/springboot/starter/adapter/springcloud/stream/rabbitmq/example/ServerAdapterSpringCloudStreamRabbitMQApplication.java new file mode 100644 index 00000000..1f0f3c74 --- /dev/null +++ b/hippo4j-example/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq-example/src/main/java/cn/hippo4j/springboot/starter/adapter/springcloud/stream/rabbitmq/example/ServerAdapterSpringCloudStreamRabbitMQApplication.java @@ -0,0 +1,52 @@ +/* + * 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.springcloud.stream.rabbitmq.example; + +import cn.hippo4j.core.enable.EnableDynamicThreadPool; +import com.alibaba.fastjson.JSON; +import lombok.extern.slf4j.Slf4j; +import org.springframework.amqp.support.AmqpHeaders; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.annotation.Bean; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; + +import java.util.function.Consumer; + +@Slf4j +@EnableDynamicThreadPool +@SpringBootApplication +public class ServerAdapterSpringCloudStreamRabbitMQApplication { + + public static void main(String[] args) { + SpringApplication.run(ServerAdapterSpringCloudStreamRabbitMQApplication.class, args); + } + + @Bean + public Consumer> demoInput() { + return message -> { + MessageHeaders headers = message.getHeaders(); + log.info("Input current thread name: {} ,{} received from partition {}", + Thread.currentThread().getName(), + JSON.toJSONString(message.getPayload()), + headers.get(AmqpHeaders.CONSUMER_QUEUE)); + }; + } + +} diff --git a/hippo4j-example/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq-example/src/main/resources/application.properties b/hippo4j-example/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq-example/src/main/resources/application.properties new file mode 100644 index 00000000..813a9f87 --- /dev/null +++ b/hippo4j-example/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq-example/src/main/resources/application.properties @@ -0,0 +1,28 @@ +server.port=8090 +logging.level.org.springframework.amqp. + +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 + +spring.rabbitmq.host=localhost +spring.rabbitmq.port=5672 +spring.rabbitmq.username=root +spring.rabbitmq.password=123456 +spring.rabbitmq.virtual-host=/ + +# Please replace the local configuration. +spring.cloud.stream.binders.defaultRabbit.type=rabbit + +# === produce === +spring.cloud.stream.bindings.demoOutput.destination=exchange-demo +spring.cloud.stream.bindings.demoOutput.content-type=application/json + +# === consume === +spring.cloud.stream.bindings.demoInput-in-0.destination=exchange-demo +spring.cloud.stream.bindings.demoInput-in-0.content-type=application/json +spring.cloud.stream.bindings.demoInput-in-0.group=myGroup +spring.cloud.stream.bindings.demoInput-in-0.consumer.concurrency=1 diff --git a/hippo4j-example/pom.xml b/hippo4j-example/pom.xml index 74e0f361..a8159637 100644 --- a/hippo4j-example/pom.xml +++ b/hippo4j-example/pom.xml @@ -19,6 +19,7 @@ hippo4j-config-zookeeper-spring-boot-starter-example hippo4j-spring-boot-starter-adapter-kafka-example hippo4j-spring-boot-starter-adapter-rabbitmq-example + hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq-example hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rocketmq-example hippo4j-spring-boot-starter-adapter-rocketmq-example diff --git a/hippo4j-spring-boot/hippo4j-spring-boot-starter-adapter/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq/pom.xml b/hippo4j-spring-boot/hippo4j-spring-boot-starter-adapter/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq/pom.xml new file mode 100644 index 00000000..c0f00029 --- /dev/null +++ b/hippo4j-spring-boot/hippo4j-spring-boot-starter-adapter/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq/pom.xml @@ -0,0 +1,51 @@ + + + 4.0.0 + + cn.hippo4j + hippo4j-spring-boot-starter-adapter + ${revision} + + hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq + hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq + + + + cn.hippo4j + hippo4j-adapter-spring-cloud-stream-rabbitmq + ${revision} + + + + + + + org.apache.maven.plugins + maven-jar-plugin + + + + ${project.artifactId} + ${project.version} + ${maven.build.timestamp} + chen.ma + + + + + + org.apache.maven.plugins + maven-javadoc-plugin + 2.10.3 + + + + jar + + + + + + + diff --git a/hippo4j-spring-boot/hippo4j-spring-boot-starter-adapter/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq/src/main/java/cn/hippo4j/springboot/starter/adapter/springcloud/stream/rabbitmq/SpringCloudStreamRabbitMQAdapterAutoConfiguration.java b/hippo4j-spring-boot/hippo4j-spring-boot-starter-adapter/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq/src/main/java/cn/hippo4j/springboot/starter/adapter/springcloud/stream/rabbitmq/SpringCloudStreamRabbitMQAdapterAutoConfiguration.java new file mode 100644 index 00000000..8a722629 --- /dev/null +++ b/hippo4j-spring-boot/hippo4j-spring-boot-starter-adapter/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq/src/main/java/cn/hippo4j/springboot/starter/adapter/springcloud/stream/rabbitmq/SpringCloudStreamRabbitMQAdapterAutoConfiguration.java @@ -0,0 +1,50 @@ +/* + * 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.springcloud.stream.rabbitmq; + +import cn.hippo4j.adapter.springcloud.stream.rabbitmq.SpringCloudStreamRabbitMQThreadPoolAdapter; +import cn.hippo4j.common.config.ApplicationContextHolder; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.cloud.stream.binder.rabbit.config.RabbitMessageChannelBinderConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * Spring cloud stream rabbitmq adapter auto configuration. + * + * @author lijianxin + */ +@Configuration(proxyBeanMethods = false) +@ConditionalOnClass(RabbitMessageChannelBinderConfiguration.class) +public class SpringCloudStreamRabbitMQAdapterAutoConfiguration { + + @Bean + @ConditionalOnMissingBean + public ApplicationContextHolder simpleApplicationContextHolder() { + return new ApplicationContextHolder(); + } + + @Bean + @SuppressWarnings("all") + @ConditionalOnProperty("spring.rabbitmq.host") + public SpringCloudStreamRabbitMQThreadPoolAdapter springCloudStreamRabbitMQThreadPoolAdapter(ApplicationContextHolder applicationContextHolder) { + return new SpringCloudStreamRabbitMQThreadPoolAdapter(); + } +} diff --git a/hippo4j-spring-boot/hippo4j-spring-boot-starter-adapter/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq/src/main/resources/META-INF/spring.factories b/hippo4j-spring-boot/hippo4j-spring-boot-starter-adapter/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq/src/main/resources/META-INF/spring.factories new file mode 100644 index 00000000..52a54b9c --- /dev/null +++ b/hippo4j-spring-boot/hippo4j-spring-boot-starter-adapter/hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq/src/main/resources/META-INF/spring.factories @@ -0,0 +1 @@ +org.springframework.boot.autoconfigure.EnableAutoConfiguration=cn.hippo4j.springboot.starter.adapter.springcloud.stream.rabbitmq.SpringCloudStreamRabbitMQAdapterAutoConfiguration diff --git a/hippo4j-spring-boot/hippo4j-spring-boot-starter-adapter/pom.xml b/hippo4j-spring-boot/hippo4j-spring-boot-starter-adapter/pom.xml index 19da4cf5..136f0878 100644 --- a/hippo4j-spring-boot/hippo4j-spring-boot-starter-adapter/pom.xml +++ b/hippo4j-spring-boot/hippo4j-spring-boot-starter-adapter/pom.xml @@ -22,6 +22,7 @@ hippo4j-spring-boot-starter-adapter-hystrix hippo4j-spring-boot-starter-adapter-spring-cloud-stream-kafka hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rocketmq + hippo4j-spring-boot-starter-adapter-spring-cloud-stream-rabbitmq hippo4j-spring-boot-starter-adapter-web diff --git a/pom.xml b/pom.xml index 54fa743c..ec01bb02 100644 --- a/pom.xml +++ b/pom.xml @@ -54,6 +54,7 @@ 9.0.55 5.1.0 2.2.6.RELEASE + 3.1.0 2.2.9.RELEASE 2.2.5.RELEASE