From 0ad0014cbf1315b55bf482750173174dc11ef72f Mon Sep 17 00:00:00 2001 From: mercybltz Date: Thu, 22 Nov 2018 13:02:33 +0800 Subject: [PATCH 01/10] Polish spring-cloud-starter-bus-rocketmq --- pom.xml | 10 +++++++ spring-cloud-alibaba-dependencies/pom.xml | 8 ++++++ spring-cloud-starter-bus-rocketmq/pom.xml | 32 +++++++++++++++++++++++ 3 files changed, 50 insertions(+) create mode 100644 spring-cloud-starter-bus-rocketmq/pom.xml diff --git a/pom.xml b/pom.xml index 1969d628..4f3a4360 100644 --- a/pom.xml +++ b/pom.xml @@ -69,6 +69,7 @@ 2.0.2.RELEASE 2.0.2.RELEASE 2.0.2.RELEASE + 2.0.0.RELEASE 4.12 3.0 @@ -98,6 +99,7 @@ spring-cloud-alicloud-oss spring-cloud-alicloud-acm spring-cloud-alicloud-ans + spring-cloud-starter-bus-rocketmq @@ -142,6 +144,14 @@ import + + org.springframework.cloud + spring-cloud-bus-dependencies + ${spring-cloud-bus.version} + pom + import + + diff --git a/spring-cloud-alibaba-dependencies/pom.xml b/spring-cloud-alibaba-dependencies/pom.xml index b663184f..5d082bf7 100644 --- a/spring-cloud-alibaba-dependencies/pom.xml +++ b/spring-cloud-alibaba-dependencies/pom.xml @@ -24,6 +24,7 @@ 4.0.1 1.0.0 2.16.0 + 4.3.1 @@ -74,6 +75,13 @@ ${nacos.version} + + + org.apache.rocketmq + rocketmq-client + ${rocketmq.version} + + com.alibaba.csp diff --git a/spring-cloud-starter-bus-rocketmq/pom.xml b/spring-cloud-starter-bus-rocketmq/pom.xml new file mode 100644 index 00000000..dacf3b82 --- /dev/null +++ b/spring-cloud-starter-bus-rocketmq/pom.xml @@ -0,0 +1,32 @@ + + + + spring-cloud-alibaba + org.springframework.cloud + 0.2.1.BUILD-SNAPSHOT + ../pom.xml + + 4.0.0 + + org.springframework.cloud + spring-cloud-starter-bus-rocketmq + + + + + + org.apache.rocketmq + rocketmq-client + + + + + org.springframework.cloud + spring-cloud-bus + + + + + \ No newline at end of file From 6e175b35d57224cf33afd8cf4487c4c1685efeee Mon Sep 17 00:00:00 2001 From: mercyblitz Date: Fri, 23 Nov 2018 16:33:24 +0800 Subject: [PATCH 02/10] Fix bugs on example --- .../rocketmq/RocketMQBusApplication.java | 59 +++++++++++++------ .../src/main/resources/application.properties | 5 -- .../src/main/resources/bootstrap.properties | 4 ++ 3 files changed, 44 insertions(+), 24 deletions(-) delete mode 100644 spring-cloud-alibaba-examples/spring-cloud-bus-rocketmq-example/src/main/resources/application.properties create mode 100644 spring-cloud-alibaba-examples/spring-cloud-bus-rocketmq-example/src/main/resources/bootstrap.properties diff --git a/spring-cloud-alibaba-examples/spring-cloud-bus-rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/rocketmq/RocketMQBusApplication.java b/spring-cloud-alibaba-examples/spring-cloud-bus-rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/rocketmq/RocketMQBusApplication.java index dfae4d83..c4a05a8b 100644 --- a/spring-cloud-alibaba-examples/spring-cloud-bus-rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/rocketmq/RocketMQBusApplication.java +++ b/spring-cloud-alibaba-examples/spring-cloud-bus-rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/rocketmq/RocketMQBusApplication.java @@ -16,15 +16,19 @@ */ package org.springframework.cloud.alibaba.cloud.examples.rocketmq; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.ApplicationRunner; -import org.springframework.boot.WebApplicationType; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.bus.event.AckRemoteApplicationEvent; import org.springframework.cloud.bus.jackson.RemoteApplicationEventScan; import org.springframework.context.ApplicationEventPublisher; -import org.springframework.context.annotation.Bean; import org.springframework.context.event.EventListener; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.bind.annotation.RestController; /** * RocketMQ Bus Spring Application @@ -32,33 +36,45 @@ import org.springframework.context.event.EventListener; * @author Mercy * @since 0.2.1 */ +@RestController @EnableAutoConfiguration @RemoteApplicationEventScan(basePackages = "org.springframework.cloud.alibaba.cloud.examples.rocketmq") public class RocketMQBusApplication { public static void main(String[] args) { new SpringApplicationBuilder(RocketMQBusApplication.class) + .properties("server.port=0") // Random server port + .properties("management.endpoints.web.exposure.include=*") // exposure includes all + .properties("spring.cloud.bus.trace.enabled=true") // Enable trace .run(args); } + @Autowired + private ApplicationEventPublisher publisher; + + @Value("${spring.cloud.bus.id}") + private String originService; + + @Value("${server.port}") + private int localServerPort; + + @Autowired + private ObjectMapper objectMapper; + /** - * Publish the {@link UserRemoteApplicationEvent} to all instances of currentService. + * Publish the {@link UserRemoteApplicationEvent} * - * @param publisher {@link ApplicationEventPublisher} - * @param currentService Current application Name - * @return {@link ApplicationRunner} instance + * @param name the user name + * @param destination the destination + * @return If published */ - @Bean - public ApplicationRunner publishEventRunner(ApplicationEventPublisher publisher, - @Value("${spring.application.name}") String currentService) { - return args -> { - User user = new User(); - user.setName("Mercy Ma"); - for (int i = 1; i < 10; i++) { - user.setId(Long.valueOf(i)); - publisher.publishEvent(new UserRemoteApplicationEvent(user, currentService, currentService + ":**")); - } - }; + @GetMapping("/bus/event/publish/user") + public boolean publish(@RequestParam String name, @RequestParam(required = false) String destination) { + User user = new User(); + user.setId(System.currentTimeMillis()); + user.setName(name); + publisher.publishEvent(new UserRemoteApplicationEvent(user, originService, destination)); + return true; } /** @@ -68,6 +84,11 @@ public class RocketMQBusApplication { */ @EventListener public void onEvent(UserRemoteApplicationEvent event) { - System.out.println("Listener on User : " + event.getUser()); + System.out.printf("Server [port : %d] listeners on %s\n", localServerPort, event.getUser()); + } + + @EventListener + public void onAckEvent(AckRemoteApplicationEvent event) throws JsonProcessingException { + System.out.printf("Server [port : %d] listeners on %s\n", localServerPort, objectMapper.writeValueAsString(event)); } } diff --git a/spring-cloud-alibaba-examples/spring-cloud-bus-rocketmq-example/src/main/resources/application.properties b/spring-cloud-alibaba-examples/spring-cloud-bus-rocketmq-example/src/main/resources/application.properties deleted file mode 100644 index 6c57bdfe..00000000 --- a/spring-cloud-alibaba-examples/spring-cloud-bus-rocketmq-example/src/main/resources/application.properties +++ /dev/null @@ -1,5 +0,0 @@ -spring.application.name=spring-cloud-bus-rocketmq-example - -management.endpoints.web.exposure.include=* - -spring.cloud.bus.trace.enabled=true \ No newline at end of file diff --git a/spring-cloud-alibaba-examples/spring-cloud-bus-rocketmq-example/src/main/resources/bootstrap.properties b/spring-cloud-alibaba-examples/spring-cloud-bus-rocketmq-example/src/main/resources/bootstrap.properties new file mode 100644 index 00000000..1b944de9 --- /dev/null +++ b/spring-cloud-alibaba-examples/spring-cloud-bus-rocketmq-example/src/main/resources/bootstrap.properties @@ -0,0 +1,4 @@ +spring.application.name=spring-cloud-bus-rocketmq-example +spring.cloud.stream.rocketmq.binder.namesrv-addr=127.0.0.1:9876 +server.port=8080 +spring.cloud.bus.id=${spring.application.name}:${server.port} \ No newline at end of file From 2ab11490dcd51c19c2405ad754fb960a040f9392 Mon Sep 17 00:00:00 2001 From: fangjian0423 Date: Tue, 27 Nov 2018 13:55:38 +0800 Subject: [PATCH 03/10] fixes #120 and do some refactor --- .../RocketMQInboundChannelAdapter.java | 4 ++-- .../metrics/ConsumerGroupInstrumentation.java | 17 ----------------- .../RocketMQBinderConfigurationProperties.java | 2 +- 3 files changed, 3 insertions(+), 20 deletions(-) diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQInboundChannelAdapter.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQInboundChannelAdapter.java index 6b22c400..84c2898e 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQInboundChannelAdapter.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQInboundChannelAdapter.java @@ -189,14 +189,14 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport { } catch (Exception e) { logger.error( - "Rocket Message hasn't been processed successfully. Caused by ", + "RocketMQ Message hasn't been processed successfully. Caused by ", e); instrumentationManager .getConsumerInstrumentation( RocketMQInboundChannelAdapter.this.destination) .markConsumedFailure(); throw new RuntimeException( - "Rocket Message hasn't been processed successfully. Caused by ", + "RocketMQ Message hasn't been processed successfully. Caused by ", e); } } diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ConsumerGroupInstrumentation.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ConsumerGroupInstrumentation.java index 83b54c3e..87f0c978 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ConsumerGroupInstrumentation.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ConsumerGroupInstrumentation.java @@ -1,7 +1,5 @@ package org.springframework.cloud.stream.binder.rocketmq.metrics; -import java.util.concurrent.atomic.AtomicBoolean; - import com.codahale.metrics.MetricRegistry; /** @@ -11,24 +9,9 @@ import com.codahale.metrics.MetricRegistry; public class ConsumerGroupInstrumentation extends Instrumentation { private MetricRegistry metricRegistry; - private AtomicBoolean delayedStart = new AtomicBoolean(false); - public ConsumerGroupInstrumentation(MetricRegistry metricRegistry, String name) { super(name); this.metricRegistry = metricRegistry; } - public void markDelayedStart() { - delayedStart.set(true); - } - - @Override - public boolean isUp() { - return started.get() || delayedStart.get(); - } - - @Override - public boolean isOutOfService() { - return !started.get() && startException == null && !delayedStart.get(); - } } diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQBinderConfigurationProperties.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQBinderConfigurationProperties.java index dd00e7d1..087d95c2 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQBinderConfigurationProperties.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQBinderConfigurationProperties.java @@ -9,7 +9,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties; @ConfigurationProperties(prefix = "spring.cloud.stream.rocketmq.binder") public class RocketMQBinderConfigurationProperties { - private String namesrvAddr; + private String namesrvAddr = "127.0.0.1:9876"; private String logLevel = "ERROR"; From 0896af32bba7df4d101a92206783a83959981c77 Mon Sep 17 00:00:00 2001 From: fangjian0423 Date: Tue, 27 Nov 2018 16:35:10 +0800 Subject: [PATCH 04/10] magic value refactor --- .../rocketmq/RocketMQBinderConstants.java | 23 +++++++++++++++++-- .../metrics/ConsumerInstrumentation.java | 12 ++++++---- .../metrics/InstrumentationManager.java | 10 +++++--- .../metrics/ProducerInstrumentation.java | 12 ++++++---- 4 files changed, 44 insertions(+), 13 deletions(-) diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQBinderConstants.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQBinderConstants.java index 3ee0335c..cd619acf 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQBinderConstants.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQBinderConstants.java @@ -5,6 +5,8 @@ package org.springframework.cloud.stream.binder.rocketmq; */ public interface RocketMQBinderConstants { + String ENDPOINT_ID = "rocketmq-binder"; + /** * Header key */ @@ -17,10 +19,27 @@ public interface RocketMQBinderConstants { String ACKNOWLEDGEMENT_KEY = "ACKNOWLEDGEMENT"; /** - * Instrumentation key + * Instrumentation */ String LASTSEND_TIMESTAMP = "lastSend.timestamp"; - String ENDPOINT_ID = "rocketmq-binder"; + interface Metrics { + interface Producer { + String PREFIX = "scs-rocketmq.producer."; + String TOTAL_SENT = "totalSent"; + String TOTAL_SENT_FAILURES = "totalSentFailures"; + String SENT_PER_SECOND = "sentPerSecond"; + String SENT_FAILURES_PER_SECOND = "sentFailuresPerSecond"; + } + + interface Consumer { + String GROUP_PREFIX = "scs-rocketmq.consumerGroup."; + String PREFIX = "scs-rocketmq.consumer."; + String TOTAL_CONSUMED = "totalConsumed"; + String CONSUMED_PER_SECOND = "consumedPerSecond"; + String TOTAL_CONSUMED_FAILURES = "totalConsumedFailures"; + String CONSUMED_FAILURES_PER_SECOND = "consumedFailuresPerSecond"; + } + } } diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ConsumerInstrumentation.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ConsumerInstrumentation.java index 76430737..ad72f4d8 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ConsumerInstrumentation.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ConsumerInstrumentation.java @@ -2,6 +2,8 @@ package org.springframework.cloud.stream.binder.rocketmq.metrics; import static com.codahale.metrics.MetricRegistry.name; +import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.Metrics.Consumer; + import com.codahale.metrics.Counter; import com.codahale.metrics.Meter; import com.codahale.metrics.MetricRegistry; @@ -19,13 +21,15 @@ public class ConsumerInstrumentation extends Instrumentation { public ConsumerInstrumentation(MetricRegistry registry, String baseMetricName) { super(baseMetricName); - this.totalConsumed = registry.counter(name(baseMetricName, "totalConsumed")); + + this.totalConsumed = registry + .counter(name(baseMetricName, Consumer.TOTAL_CONSUMED)); this.consumedPerSecond = registry - .meter(name(baseMetricName, "consumedPerSecond")); + .meter(name(baseMetricName, Consumer.CONSUMED_PER_SECOND)); this.totalConsumedFailures = registry - .counter(name(baseMetricName, "totalConsumedFailures")); + .counter(name(baseMetricName, Consumer.TOTAL_CONSUMED_FAILURES)); this.consumedFailuresPerSecond = registry - .meter(name(baseMetricName, "consumedFailuresPerSecond")); + .meter(name(baseMetricName, Consumer.CONSUMED_FAILURES_PER_SECOND)); } public void markConsumed() { diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/InstrumentationManager.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/InstrumentationManager.java index 601356f2..599401c3 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/InstrumentationManager.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/InstrumentationManager.java @@ -5,6 +5,9 @@ import java.util.Map; import java.util.Set; import java.util.stream.Collectors; +import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.Metrics.Consumer; +import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.Metrics.Producer; + import com.codahale.metrics.MetricRegistry; /** @@ -27,21 +30,22 @@ public class InstrumentationManager { } public ProducerInstrumentation getProducerInstrumentation(String destination) { - String key = "scs-rocketmq.producer." + destination; + + String key = Producer.PREFIX + destination; producerInstrumentations.putIfAbsent(key, new ProducerInstrumentation(metricRegistry, key)); return producerInstrumentations.get(key); } public ConsumerInstrumentation getConsumerInstrumentation(String destination) { - String key = "scs-rocketmq.consumer." + destination; + String key = Consumer.PREFIX + destination; consumeInstrumentations.putIfAbsent(key, new ConsumerInstrumentation(metricRegistry, key)); return consumeInstrumentations.get(key); } public ConsumerGroupInstrumentation getConsumerGroupInstrumentation(String group) { - String key = "scs-rocketmq.consumerGroup." + group; + String key = Consumer.GROUP_PREFIX + group; consumerGroupsInstrumentations.putIfAbsent(key, new ConsumerGroupInstrumentation(metricRegistry, key)); return consumerGroupsInstrumentations.get(key); diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ProducerInstrumentation.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ProducerInstrumentation.java index 1db95ccf..68c9c0ee 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ProducerInstrumentation.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ProducerInstrumentation.java @@ -2,6 +2,8 @@ package org.springframework.cloud.stream.binder.rocketmq.metrics; import static com.codahale.metrics.MetricRegistry.name; +import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.Metrics.Producer; + import com.codahale.metrics.Counter; import com.codahale.metrics.Meter; import com.codahale.metrics.MetricRegistry; @@ -19,12 +21,14 @@ public class ProducerInstrumentation extends Instrumentation { public ProducerInstrumentation(MetricRegistry registry, String baseMetricName) { super(baseMetricName); - this.totalSent = registry.counter(name(baseMetricName, "totalSent")); + + this.totalSent = registry.counter(name(baseMetricName, Producer.TOTAL_SENT)); this.totalSentFailures = registry - .counter(name(baseMetricName, "totalSentFailures")); - this.sentPerSecond = registry.meter(name(baseMetricName, "sentPerSecond")); + .counter(name(baseMetricName, Producer.TOTAL_SENT_FAILURES)); + this.sentPerSecond = registry + .meter(name(baseMetricName, Producer.SENT_PER_SECOND)); this.sentFailuresPerSecond = registry - .meter(name(baseMetricName, "sentFailuresPerSecond")); + .meter(name(baseMetricName, Producer.SENT_FAILURES_PER_SECOND)); } public void markSent() { From 7d039f47425f5d32c647112d6d6a5524cbdb7f53 Mon Sep 17 00:00:00 2001 From: fangjian0423 Date: Wed, 28 Nov 2018 17:13:59 +0800 Subject: [PATCH 05/10] make metrics-core as provided dependency and refactor --- pom.xml | 1 - .../rocketmq-example/pom.xml | 6 ++ spring-cloud-stream-binder-rocketmq/pom.xml | 11 ++- .../RocketMQMessageChannelBinder.java | 6 +- .../actuator/RocketMQBinderEndpoint.java | 28 +++--- .../RocketMQBinderHealthIndicator.java | 42 ++++---- .../RocketMQBinderAutoConfiguration.java | 7 +- ...cketMQBinderEndpointAutoConfiguration.java | 15 +-- .../rocketmq/consuming/ConsumersManager.java | 30 ++++-- .../RocketMQInboundChannelAdapter.java | 99 +++++++++---------- .../integration/RocketMQMessageHandler.java | 30 +++--- .../metrics/InstrumentationManager.java | 18 ++-- 12 files changed, 161 insertions(+), 132 deletions(-) diff --git a/pom.xml b/pom.xml index a2a7aa0f..b0532ec3 100644 --- a/pom.xml +++ b/pom.xml @@ -100,7 +100,6 @@ spring-cloud-alicloud-oss spring-cloud-alicloud-acm spring-cloud-alicloud-ans - spring-cloud-starter-bus-rocketmq diff --git a/spring-cloud-alibaba-examples/rocketmq-example/pom.xml b/spring-cloud-alibaba-examples/rocketmq-example/pom.xml index bec837c3..618fb069 100644 --- a/spring-cloud-alibaba-examples/rocketmq-example/pom.xml +++ b/spring-cloud-alibaba-examples/rocketmq-example/pom.xml @@ -31,6 +31,12 @@ org.springframework.boot spring-boot-starter-actuator + + + io.dropwizard.metrics + metrics-core + + diff --git a/spring-cloud-stream-binder-rocketmq/pom.xml b/spring-cloud-stream-binder-rocketmq/pom.xml index 438d27d4..9449c430 100644 --- a/spring-cloud-stream-binder-rocketmq/pom.xml +++ b/spring-cloud-stream-binder-rocketmq/pom.xml @@ -22,13 +22,16 @@ - io.dropwizard.metrics - metrics-core + org.apache.rocketmq + rocketmq-client - org.apache.rocketmq - rocketmq-client + io.dropwizard.metrics + metrics-core + 4.0.3 + provided + true diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java index cb3d5cad..a43b189c 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java @@ -34,7 +34,6 @@ public class RocketMQMessageChannelBinder extends .getLogger(RocketMQMessageChannelBinder.class); private final RocketMQExtendedBindingProperties extendedBindingProperties; - private final RocketMQTopicProvisioner rocketTopicProvisioner; private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties; private final InstrumentationManager instrumentationManager; private final ConsumersManager consumersManager; @@ -47,7 +46,6 @@ public class RocketMQMessageChannelBinder extends super(null, provisioningProvider); this.consumersManager = consumersManager; this.extendedBindingProperties = extendedBindingProperties; - this.rocketTopicProvisioner = provisioningProvider; this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties; this.instrumentationManager = instrumentationManager; } @@ -63,7 +61,7 @@ public class RocketMQMessageChannelBinder extends } else { throw new RuntimeException("Binding for channel " + destination.getName() - + "has been disabled, message can't be delivered"); + + " has been disabled, message can't be delivered"); } } @@ -74,7 +72,7 @@ public class RocketMQMessageChannelBinder extends throws Exception { if (group == null || "".equals(group)) { throw new RuntimeException( - "'group' must be configured for channel + " + destination.getName()); + "'group must be configured for channel + " + destination.getName()); } RocketMQInboundChannelAdapter rocketInboundChannelAdapter = new RocketMQInboundChannelAdapter( diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/actuator/RocketMQBinderEndpoint.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/actuator/RocketMQBinderEndpoint.java index 5a7f57f5..5c4f9e67 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/actuator/RocketMQBinderEndpoint.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/actuator/RocketMQBinderEndpoint.java @@ -4,12 +4,11 @@ import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderCon import java.util.HashMap; import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.actuate.endpoint.annotation.Endpoint; import org.springframework.boot.actuate.endpoint.annotation.ReadOperation; - -import com.codahale.metrics.MetricRegistry; +import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationManager; /** * @author Timur Valiev @@ -18,23 +17,22 @@ import com.codahale.metrics.MetricRegistry; @Endpoint(id = ENDPOINT_ID) public class RocketMQBinderEndpoint { - private MetricRegistry metricRegistry = new MetricRegistry(); - private Map runtime = new ConcurrentHashMap<>(); + @Autowired(required = false) + private InstrumentationManager instrumentationManager; @ReadOperation public Map invoke() { Map result = new HashMap<>(); - result.put("metrics", metricRegistry().getMetrics()); - result.put("runtime", runtime()); + if (instrumentationManager != null) { + result.put("metrics", + instrumentationManager.getMetricRegistry().getMetrics()); + result.put("runtime", instrumentationManager.getRuntime()); + } + else { + result.put("warning", + "please add metrics-core dependency, we use it for metrics"); + } return result; } - public MetricRegistry metricRegistry() { - return metricRegistry; - } - - public Map runtime() { - return runtime; - } - } diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/actuator/RocketMQBinderHealthIndicator.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/actuator/RocketMQBinderHealthIndicator.java index 82f4ab5f..b517a6f7 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/actuator/RocketMQBinderHealthIndicator.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/actuator/RocketMQBinderHealthIndicator.java @@ -1,5 +1,6 @@ package org.springframework.cloud.stream.binder.rocketmq.actuator; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.actuate.health.AbstractHealthIndicator; import org.springframework.boot.actuate.health.Health; import org.springframework.cloud.stream.binder.rocketmq.metrics.Instrumentation; @@ -11,28 +12,33 @@ import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationM */ public class RocketMQBinderHealthIndicator extends AbstractHealthIndicator { - private final InstrumentationManager instrumentationManager; - - public RocketMQBinderHealthIndicator(InstrumentationManager instrumentationManager) { - this.instrumentationManager = instrumentationManager; - } + @Autowired(required = false) + private InstrumentationManager instrumentationManager; @Override protected void doHealthCheck(Health.Builder builder) throws Exception { - if (instrumentationManager.getHealthInstrumentations().stream() - .allMatch(Instrumentation::isUp)) { - builder.up(); - return; + if (instrumentationManager != null) { + if (instrumentationManager.getHealthInstrumentations().stream() + .allMatch(Instrumentation::isUp)) { + builder.up(); + return; + } + if (instrumentationManager.getHealthInstrumentations().stream() + .allMatch(Instrumentation::isOutOfService)) { + builder.outOfService(); + return; + } + builder.down(); + instrumentationManager.getHealthInstrumentations().stream() + .filter(instrumentation -> !instrumentation.isStarted()) + .forEach(instrumentation1 -> builder + .withException(instrumentation1.getStartException())); } - if (instrumentationManager.getHealthInstrumentations().stream() - .allMatch(Instrumentation::isOutOfService)) { - builder.outOfService(); - return; + else { + builder.down(); + builder.withDetail("warning", + "please add metrics-core dependency, we use it for metrics"); } - builder.down(); - instrumentationManager.getHealthInstrumentations().stream() - .filter(instrumentation -> !instrumentation.isStarted()) - .forEach(instrumentation1 -> builder - .withException(instrumentation1.getStartException())); + } } diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderAutoConfiguration.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderAutoConfiguration.java index 713edb5c..dda8d439 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderAutoConfiguration.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderAutoConfiguration.java @@ -25,6 +25,9 @@ public class RocketMQBinderAutoConfiguration { private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties; + @Autowired(required = false) + private InstrumentationManager instrumentationManager; + @Autowired public RocketMQBinderAutoConfiguration( RocketMQExtendedBindingProperties extendedBindingProperties, @@ -43,7 +46,6 @@ public class RocketMQBinderAutoConfiguration { @Bean public RocketMQMessageChannelBinder rocketMessageChannelBinder( RocketMQTopicProvisioner provisioningProvider, - InstrumentationManager instrumentationManager, ConsumersManager consumersManager) { RocketMQMessageChannelBinder binder = new RocketMQMessageChannelBinder( consumersManager, extendedBindingProperties, provisioningProvider, @@ -52,8 +54,7 @@ public class RocketMQBinderAutoConfiguration { } @Bean - public ConsumersManager consumersManager( - InstrumentationManager instrumentationManager) { + public ConsumersManager consumersManager() { return new ConsumersManager(instrumentationManager, rocketBinderConfigurationProperties); } diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderEndpointAutoConfiguration.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderEndpointAutoConfiguration.java index 1b0bd0f5..7e40cd5b 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderEndpointAutoConfiguration.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderEndpointAutoConfiguration.java @@ -1,7 +1,9 @@ package org.springframework.cloud.stream.binder.rocketmq.config; import org.springframework.boot.actuate.autoconfigure.endpoint.EndpointAutoConfiguration; +import org.springframework.boot.actuate.endpoint.annotation.Endpoint; import org.springframework.boot.autoconfigure.AutoConfigureAfter; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.cloud.stream.binder.rocketmq.actuator.RocketMQBinderEndpoint; import org.springframework.cloud.stream.binder.rocketmq.actuator.RocketMQBinderHealthIndicator; import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationManager; @@ -13,6 +15,7 @@ import org.springframework.context.annotation.Configuration; */ @Configuration @AutoConfigureAfter(EndpointAutoConfiguration.class) +@ConditionalOnClass(Endpoint.class) public class RocketMQBinderEndpointAutoConfiguration { @Bean @@ -21,16 +24,14 @@ public class RocketMQBinderEndpointAutoConfiguration { } @Bean - public RocketMQBinderHealthIndicator rocketBinderHealthIndicator( - InstrumentationManager instrumentationManager) { - return new RocketMQBinderHealthIndicator(instrumentationManager); + public RocketMQBinderHealthIndicator rocketBinderHealthIndicator() { + return new RocketMQBinderHealthIndicator(); } @Bean - public InstrumentationManager instrumentationManager( - RocketMQBinderEndpoint rocketBinderEndpoint) { - return new InstrumentationManager(rocketBinderEndpoint.metricRegistry(), - rocketBinderEndpoint.runtime()); + @ConditionalOnClass(name = "com.codahale.metrics.Counter") + public InstrumentationManager instrumentationManager() { + return new InstrumentationManager(); } } diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/consuming/ConsumersManager.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/consuming/ConsumersManager.java index 4184e3f4..2dd425b0 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/consuming/ConsumersManager.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/consuming/ConsumersManager.java @@ -3,6 +3,7 @@ package org.springframework.cloud.stream.binder.rocketmq.consuming; import java.util.AbstractMap; import java.util.HashMap; import java.util.Map; +import java.util.Optional; import java.util.Set; import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; @@ -27,9 +28,10 @@ public class ConsumersManager { private final Map consumerGroups = new HashMap<>(); private final Map started = new HashMap<>(); private final Map, ExtendedConsumerProperties> propertiesMap = new HashMap<>(); - private final InstrumentationManager instrumentationManager; private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties; + private InstrumentationManager instrumentationManager; + public ConsumersManager(InstrumentationManager instrumentationManager, RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties) { this.instrumentationManager = instrumentationManager; @@ -41,9 +43,12 @@ public class ConsumersManager { ExtendedConsumerProperties consumerProperties) { propertiesMap.put(new AbstractMap.SimpleEntry<>(group, topic), consumerProperties); - ConsumerGroupInstrumentation instrumentation = instrumentationManager - .getConsumerGroupInstrumentation(group); - instrumentationManager.addHealthInstrumentation(instrumentation); + + Optional.ofNullable(instrumentationManager).ifPresent(manager -> { + ConsumerGroupInstrumentation instrumentation = manager + .getConsumerGroupInstrumentation(group); + instrumentationManager.addHealthInstrumentation(instrumentation); + }); if (consumerGroups.containsKey(group)) { return consumerGroups.get(group); @@ -87,16 +92,23 @@ public class ConsumersManager { if (started.get(group)) { return; } - ConsumerGroupInstrumentation groupInstrumentation = instrumentationManager - .getConsumerGroupInstrumentation(group); - instrumentationManager.addHealthInstrumentation(groupInstrumentation); + + ConsumerGroupInstrumentation groupInstrumentation = null; + if (Optional.ofNullable(instrumentationManager).isPresent()) { + groupInstrumentation = instrumentationManager + .getConsumerGroupInstrumentation(group); + instrumentationManager.addHealthInstrumentation(groupInstrumentation); + } + try { consumerGroups.get(group).start(); started.put(group, true); - groupInstrumentation.markStartedSuccessfully(); + Optional.ofNullable(groupInstrumentation) + .ifPresent(g -> g.markStartedSuccessfully()); } catch (MQClientException e) { - groupInstrumentation.markStartFailed(e); + Optional.ofNullable(groupInstrumentation) + .ifPresent(g -> g.markStartFailed(e)); logger.error("RocketMQ Consumer hasn't been started. Caused by " + e.getErrorMessage(), e); throw e; diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQInboundChannelAdapter.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQInboundChannelAdapter.java index 84c2898e..1831dd7b 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQInboundChannelAdapter.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQInboundChannelAdapter.java @@ -4,6 +4,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.HashSet; import java.util.List; +import java.util.Optional; import java.util.Set; import java.util.stream.Collectors; @@ -48,20 +49,20 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport { private ConsumerInstrumentation consumerInstrumentation; + private InstrumentationManager instrumentationManager; + + private RetryTemplate retryTemplate; + + private RecoveryCallback recoveryCallback; + private final ExtendedConsumerProperties consumerProperties; private final String destination; private final String group; - private final InstrumentationManager instrumentationManager; - private final ConsumersManager consumersManager; - private RetryTemplate retryTemplate; - - private RecoveryCallback recoveryCallback; - public RocketMQInboundChannelAdapter(ConsumersManager consumersManager, ExtendedConsumerProperties consumerProperties, String destination, String group, @@ -75,21 +76,20 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport { @Override protected void doStart() { - if (!consumerProperties.getExtension().getEnabled()) { + if (consumerProperties == null + || !consumerProperties.getExtension().getEnabled()) { return; } - String tags = consumerProperties == null ? null - : consumerProperties.getExtension().getTags(); - Boolean isOrderly = consumerProperties == null ? false - : consumerProperties.getExtension().getOrderly(); + String tags = consumerProperties.getExtension().getTags(); + Boolean isOrderly = consumerProperties.getExtension().getOrderly(); DefaultMQPushConsumer consumer = consumersManager.getOrCreateConsumer(group, destination, consumerProperties); final CloudStreamMessageListener listener = isOrderly - ? new CloudStreamMessageListenerOrderly(instrumentationManager) - : new CloudStreamMessageListenerConcurrently(instrumentationManager); + ? new CloudStreamMessageListenerOrderly() + : new CloudStreamMessageListenerConcurrently(); if (retryTemplate != null) { retryTemplate.registerListener(listener); @@ -99,9 +99,10 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport { : Arrays.stream(tags.split("\\|\\|")).map(String::trim) .collect(Collectors.toSet()); - consumerInstrumentation = instrumentationManager - .getConsumerInstrumentation(destination); - instrumentationManager.addHealthInstrumentation(consumerInstrumentation); + Optional.ofNullable(instrumentationManager).ifPresent(manager -> { + consumerInstrumentation = manager.getConsumerInstrumentation(destination); + manager.addHealthInstrumentation(consumerInstrumentation); + }); try { if (!StringUtils.isEmpty(consumerProperties.getExtension().getSql())) { @@ -111,10 +112,12 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport { else { consumer.subscribe(destination, String.join(" || ", tagsSet)); } - consumerInstrumentation.markStartedSuccessfully(); + Optional.ofNullable(consumerInstrumentation) + .ifPresent(c -> c.markStartedSuccessfully()); } catch (MQClientException e) { - consumerInstrumentation.markStartFailed(e); + Optional.ofNullable(consumerInstrumentation) + .ifPresent(c -> c.markStartFailed(e)); logger.error("RocketMQ Consumer hasn't been subscribed. Caused by " + e.getErrorMessage(), e); throw new RuntimeException("RocketMQ Consumer hasn't been subscribed.", e); @@ -148,12 +151,6 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport { protected class CloudStreamMessageListener implements MessageListener, RetryListener { - private final InstrumentationManager instrumentationManager; - - CloudStreamMessageListener(InstrumentationManager instrumentationManager) { - this.instrumentationManager = instrumentationManager; - } - Acknowledgement consumeMessage(final List msgs) { boolean enableRetry = RocketMQInboundChannelAdapter.this.retryTemplate != null; try { @@ -180,10 +177,13 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport { } else { Acknowledgement result = doSendMsgs(msgs, null); - instrumentationManager - .getConsumerInstrumentation( - RocketMQInboundChannelAdapter.this.destination) - .markConsumed(); + Optional.ofNullable( + RocketMQInboundChannelAdapter.this.instrumentationManager) + .ifPresent(manager -> { + manager.getConsumerInstrumentation( + RocketMQInboundChannelAdapter.this.destination) + .markConsumed(); + }); return result; } } @@ -191,10 +191,13 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport { logger.error( "RocketMQ Message hasn't been processed successfully. Caused by ", e); - instrumentationManager - .getConsumerInstrumentation( - RocketMQInboundChannelAdapter.this.destination) - .markConsumedFailure(); + Optional.ofNullable( + RocketMQInboundChannelAdapter.this.instrumentationManager) + .ifPresent(manager -> { + manager.getConsumerInstrumentation( + RocketMQInboundChannelAdapter.this.destination) + .markConsumedFailure(); + }); throw new RuntimeException( "RocketMQ Message hasn't been processed successfully. Caused by ", e); @@ -232,16 +235,22 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport { public void close(RetryContext context, RetryCallback callback, Throwable throwable) { if (throwable != null) { - instrumentationManager - .getConsumerInstrumentation( - RocketMQInboundChannelAdapter.this.destination) - .markConsumedFailure(); + Optional.ofNullable( + RocketMQInboundChannelAdapter.this.instrumentationManager) + .ifPresent(manager -> { + manager.getConsumerInstrumentation( + RocketMQInboundChannelAdapter.this.destination) + .markConsumedFailure(); + }); } else { - instrumentationManager - .getConsumerInstrumentation( - RocketMQInboundChannelAdapter.this.destination) - .markConsumed(); + Optional.ofNullable( + RocketMQInboundChannelAdapter.this.instrumentationManager) + .ifPresent(manager -> { + manager.getConsumerInstrumentation( + RocketMQInboundChannelAdapter.this.destination) + .markConsumed(); + }); } } @@ -254,11 +263,6 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport { protected class CloudStreamMessageListenerConcurrently extends CloudStreamMessageListener implements MessageListenerConcurrently { - public CloudStreamMessageListenerConcurrently( - InstrumentationManager instrumentationManager) { - super(instrumentationManager); - } - @Override public ConsumeConcurrentlyStatus consumeMessage(final List msgs, ConsumeConcurrentlyContext context) { @@ -272,11 +276,6 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport { protected class CloudStreamMessageListenerOrderly extends CloudStreamMessageListener implements MessageListenerOrderly { - public CloudStreamMessageListenerOrderly( - InstrumentationManager instrumentationManager) { - super(instrumentationManager); - } - @Override public ConsumeOrderlyStatus consumeMessage(List msgs, ConsumeOrderlyContext context) { diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQMessageHandler.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQMessageHandler.java index fb92aba7..1af26139 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQMessageHandler.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQMessageHandler.java @@ -2,6 +2,7 @@ package org.springframework.cloud.stream.binder.rocketmq.integration; import java.time.Instant; import java.util.Map; +import java.util.Optional; import org.apache.rocketmq.client.exception.MQBrokerException; import org.apache.rocketmq.client.exception.MQClientException; @@ -30,14 +31,14 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li private ProducerInstrumentation producerInstrumentation; + private InstrumentationManager instrumentationManager; + private final RocketMQProducerProperties producerProperties; private final String destination; private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties; - private final InstrumentationManager instrumentationManager; - protected volatile boolean running = false; public RocketMQMessageHandler(String destination, @@ -54,9 +55,10 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li public void start() { producer = new DefaultMQProducer(destination); - producerInstrumentation = instrumentationManager - .getProducerInstrumentation(destination); - instrumentationManager.addHealthInstrumentation(producerInstrumentation); + Optional.ofNullable(instrumentationManager).ifPresent(manager -> { + producerInstrumentation = manager.getProducerInstrumentation(destination); + manager.addHealthInstrumentation(producerInstrumentation); + }); producer.setNamesrvAddr(rocketBinderConfigurationProperties.getNamesrvAddr()); @@ -66,10 +68,12 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li try { producer.start(); - producerInstrumentation.markStartedSuccessfully(); + Optional.ofNullable(producerInstrumentation) + .ifPresent(p -> p.markStartedSuccessfully()); } catch (MQClientException e) { - producerInstrumentation.markStartFailed(e); + Optional.ofNullable(producerInstrumentation) + .ifPresent(p -> p.markStartFailed(e)); logger.error( "RocketMQ Message hasn't been sent. Caused by " + e.getMessage()); throw new MessagingException(e.getMessage(), e); @@ -127,14 +131,16 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li RocketMQMessageHeaderAccessor.putSendResult((MutableMessage) message, sendRes); } - instrumentationManager.getRuntime().put( - RocketMQBinderConstants.LASTSEND_TIMESTAMP, - Instant.now().toEpochMilli()); - producerInstrumentation.markSent(); + Optional.ofNullable(instrumentationManager).ifPresent(manager -> { + manager.getRuntime().put(RocketMQBinderConstants.LASTSEND_TIMESTAMP, + Instant.now().toEpochMilli()); + }); + Optional.ofNullable(producerInstrumentation).ifPresent(p -> p.markSent()); } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException | UnsupportedOperationException e) { - producerInstrumentation.markSentFailure(); + Optional.ofNullable(producerInstrumentation) + .ifPresent(p -> p.markSentFailure()); logger.error( "RocketMQ Message hasn't been sent. Caused by " + e.getMessage()); throw new MessagingException(e.getMessage(), e); diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/InstrumentationManager.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/InstrumentationManager.java index 599401c3..811ba01d 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/InstrumentationManager.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/InstrumentationManager.java @@ -3,6 +3,7 @@ package org.springframework.cloud.stream.binder.rocketmq.metrics; import java.util.HashMap; import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.stream.Collectors; import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.Metrics.Consumer; @@ -15,22 +16,17 @@ import com.codahale.metrics.MetricRegistry; * @author Jim */ public class InstrumentationManager { - private final MetricRegistry metricRegistry; - private final Map runtime; + + private final MetricRegistry metricRegistry = new MetricRegistry(); + private final Map runtime = new ConcurrentHashMap<>(); + private final Map producerInstrumentations = new HashMap<>(); private final Map consumeInstrumentations = new HashMap<>(); private final Map consumerGroupsInstrumentations = new HashMap<>(); private final Map healthInstrumentations = new HashMap<>(); - public InstrumentationManager(MetricRegistry metricRegistry, - Map runtime) { - this.metricRegistry = metricRegistry; - this.runtime = runtime; - } - public ProducerInstrumentation getProducerInstrumentation(String destination) { - String key = Producer.PREFIX + destination; producerInstrumentations.putIfAbsent(key, new ProducerInstrumentation(metricRegistry, key)); @@ -63,4 +59,8 @@ public class InstrumentationManager { public Map getRuntime() { return runtime; } + + public MetricRegistry getMetricRegistry() { + return metricRegistry; + } } From baea051027775dd1c454702ba81adb42f5dc6e48 Mon Sep 17 00:00:00 2001 From: fangjian0423 Date: Wed, 28 Nov 2018 17:14:25 +0800 Subject: [PATCH 06/10] update rocketmq example --- .../rocketmq-example/readme-zh.md | 14 +++++++++--- .../cloud/alibaba/cloud/examples/Foo.java | 22 +++++++++---------- .../src/main/resources/application.properties | 4 ++-- 3 files changed, 24 insertions(+), 16 deletions(-) diff --git a/spring-cloud-alibaba-examples/rocketmq-example/readme-zh.md b/spring-cloud-alibaba-examples/rocketmq-example/readme-zh.md index f8f4e4ee..bd724a53 100644 --- a/spring-cloud-alibaba-examples/rocketmq-example/readme-zh.md +++ b/spring-cloud-alibaba-examples/rocketmq-example/readme-zh.md @@ -122,12 +122,12 @@ spring.cloud.stream.bindings.output.destination=test-topic spring.cloud.stream.bindings.output.content-type=application/json spring.cloud.stream.bindings.input1.destination=test-topic -spring.cloud.stream.bindings.input1.content-type=application/json +spring.cloud.stream.bindings.input1.content-type=text/plain spring.cloud.stream.bindings.input1.group=test-group1 spring.cloud.stream.rocketmq.bindings.input1.consumer.orderly=true spring.cloud.stream.bindings.input2.destination=test-topic -spring.cloud.stream.bindings.input2.content-type=application/json +spring.cloud.stream.bindings.input2.content-type=text/plain spring.cloud.stream.bindings.input2.group=test-group2 spring.cloud.stream.rocketmq.bindings.input2.consumer.orderly=false spring.cloud.stream.rocketmq.bindings.input2.consumer.tags=tagStr @@ -195,7 +195,7 @@ Spring Boot 应用支持通过 Endpoint 来暴露相关信息，RocketMQ Stream * Spring Boot 1.x 中添加配置 `management.security.enabled=false` * Spring Boot 2.x 中添加配置 `management.endpoints.web.exposure.include=*` -Spring Boot 1.x 可以通过访问 http://127.0.0.1:18083/rocketmq-binder 来查看 RocketMQ Binder Endpoint 的信息。Spring Boot 2.x 可以通过访问 http://127.0.0.1:28081/acutator/rocketmq-binder 来访问。 +Spring Boot 1.x 可以通过访问 http://127.0.0.1:18083/rocketmq_binder 来查看 RocketMQ Binder Endpoint 的信息。Spring Boot 2.x 可以通过访问 http://127.0.0.1:28081/actuator/rocketmq-binder 来访问。 这里会统计消息最后一次发送的数据，消息发送成功或失败的次数，消息消费成功或失败的次数等数据。 @@ -249,6 +249,14 @@ Spring Boot 1.x 可以通过访问 http://127.0.0.1:18083/rocketmq-binder 来查 } ``` +注意：要想查看统计数据需要在pom里加上 [metrics-core依赖](https://mvnrepository.com/artifact/io.dropwizard.metrics/metrics-core) 。如若不加，endpoint将会显示warning信息而不会显示统计信息： + +```json +{ + "warning": "please add metrics-core dependency, we use it to metrics" +} +``` + ## More RocketMQ 是一款功能强大的分布式消息系统，广泛应用于多个领域，包括异步通信解耦、企业解决方案、金融支付、电信、电子商务、快递物流、广告营销、社交、即时通信、移动应用、手游、视频、物联网、车联网等。 diff --git a/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/Foo.java b/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/Foo.java index 652400a2..e98b6a10 100644 --- a/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/Foo.java +++ b/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/Foo.java @@ -6,14 +6,14 @@ package org.springframework.cloud.alibaba.cloud.examples; public class Foo { private int id; - private String tag; + private String bar; public Foo() { } - public Foo(int id, String tag) { + public Foo(int id, String bar) { this.id = id; - this.tag = tag; + this.bar = bar; } public int getId() { @@ -24,16 +24,16 @@ public class Foo { this.id = id; } - public String getTag() { - return tag; - } + public String getBar() { + return bar; + } - public void setTag(String tag) { - this.tag = tag; - } + public void setBar(String bar) { + this.bar = bar; + } - @Override + @Override public String toString() { - return "Foo{" + "id=" + id + ", tag='" + tag + '\'' + '}'; + return "Foo{" + "id=" + id + ", bar='" + bar + '\'' + '}'; } } diff --git a/spring-cloud-alibaba-examples/rocketmq-example/src/main/resources/application.properties b/spring-cloud-alibaba-examples/rocketmq-example/src/main/resources/application.properties index e936d25c..c92dd24a 100644 --- a/spring-cloud-alibaba-examples/rocketmq-example/src/main/resources/application.properties +++ b/spring-cloud-alibaba-examples/rocketmq-example/src/main/resources/application.properties @@ -6,13 +6,13 @@ spring.cloud.stream.bindings.output.destination=test-topic spring.cloud.stream.bindings.output.content-type=application/json spring.cloud.stream.bindings.input1.destination=test-topic -spring.cloud.stream.bindings.input1.content-type=application/json +spring.cloud.stream.bindings.input1.content-type=text/plain spring.cloud.stream.bindings.input1.group=test-group1 spring.cloud.stream.rocketmq.bindings.input1.consumer.orderly=true spring.cloud.stream.bindings.input1.consumer.maxAttempts=1 spring.cloud.stream.bindings.input2.destination=test-topic -spring.cloud.stream.bindings.input2.content-type=application/json +spring.cloud.stream.bindings.input2.content-type=text/plain spring.cloud.stream.bindings.input2.group=test-group2 spring.cloud.stream.rocketmq.bindings.input2.consumer.orderly=false spring.cloud.stream.rocketmq.bindings.input2.consumer.tags=tagStr From d18e5c3a99ee6e65afef4a08b5c0f62393db4813 Mon Sep 17 00:00:00 2001 From: pbting <314226532@qq.com> Date: Wed, 28 Nov 2018 17:17:22 +0800 Subject: [PATCH 07/10] Update nacos-config.adoc --- .../src/main/asciidoc-zh/nacos-config.adoc | 22 +++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/spring-cloud-alibaba-docs/src/main/asciidoc-zh/nacos-config.adoc b/spring-cloud-alibaba-docs/src/main/asciidoc-zh/nacos-config.adoc index d703b91b..09288c47 100644 --- a/spring-cloud-alibaba-docs/src/main/asciidoc-zh/nacos-config.adoc +++ b/spring-cloud-alibaba-docs/src/main/asciidoc-zh/nacos-config.adoc @@ -272,3 +272,25 @@ in product-env enviroment; user name :nacos-config-yaml-update; age: 68 ---- NOTE: 此案例中我们通过 `spring.profiles.active=****` 的方式写死在配置文件中，而在真正的项目实施过程中这个变量的值是需要不同环境而有不同的值。这个时候通常的做法是通过 `-Dspring.profiles.active=****` 参数指定其配置来达到环境间灵活的切换。 + +=== 支持自定义 namespace 的配置 + +在没有明确指定 `${spring.cloud.nacos.config.namespace}` 配置的情况下， 默认使用的是 Nacos 上 Public 这个namespae。在我们日常业务的开发过程中，经常需要自定义自己的 namespace，那这个时候可以通过以下配置来实现： + +[source,properties] +---- +spring.cloud.nacos.config.namespace=b3404bc0-d7dc-4855-b519-570ed34b62d7 +---- + +NOTE: 该配置必须放在 bootstrap.properties 文件中。此外 `spring.cloud.nacos.config.namespace` 的值是 namespace 对应的 id，id 值可以在 Nacos 的控制台获取。并且在添加配置时注意不要选择其他的 namespae，否则将会导致读取不到正确的配置。 + +=== 支持自定义 Group 的配置 + +在没有明确指定 `${spring.cloud.nacos.config.group}` 配置的情况下， 默认使用的是 DEFAULT_GROUP 。在我们日常业务的开发过程中，经常需要自定义自己的 Group，那这个时候可以通过以下配置来实现： + +[source,properties] +---- +spring.cloud.nacos.config.group=DEVELOP_GROUP +---- + +NOTE: 该配置必须放在 bootstrap.properties 文件中。并且在添加配置时 Group 的值一定要和 `spring.cloud.nacos.config.group` 的配置值一致。 From 9d59a0e2e535b6715509c08309c2dbf5f61e96af Mon Sep 17 00:00:00 2001 From: fangjian0423 Date: Thu, 29 Nov 2018 14:30:36 +0800 Subject: [PATCH 08/10] add apache license for rocketmq binder --- .../rocketmq/RocketMQBinderConstants.java | 16 +++ .../RocketMQMessageChannelBinder.java | 16 +++ .../RocketMQMessageHeaderAccessor.java | 16 +++ .../actuator/RocketMQBinderEndpoint.java | 16 +++ .../RocketMQBinderHealthIndicator.java | 16 +++ .../RocketMQBinderAutoConfiguration.java | 16 +++ ...cketMQBinderEndpointAutoConfiguration.java | 16 +++ .../rocketmq/consuming/Acknowledgement.java | 16 +++ .../rocketmq/consuming/ConsumersManager.java | 16 +++ .../RocketMQInboundChannelAdapter.java | 16 +++ .../integration/RocketMQMessageHandler.java | 16 +++ .../metrics/ConsumerGroupInstrumentation.java | 16 +++ .../metrics/ConsumerInstrumentation.java | 16 +++ .../rocketmq/metrics/Instrumentation.java | 16 +++ .../metrics/InstrumentationManager.java | 16 +++ .../metrics/ProducerInstrumentation.java | 16 +++ ...RocketMQBinderConfigurationProperties.java | 16 +++ .../properties/RocketMQBindingProperties.java | 16 +++ .../RocketMQConsumerProperties.java | 109 ++++++++++-------- .../RocketMQExtendedBindingProperties.java | 16 +++ .../RocketMQProducerProperties.java | 16 +++ .../RocketMQTopicProvisioner.java | 16 +++ 22 files changed, 398 insertions(+), 47 deletions(-) diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQBinderConstants.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQBinderConstants.java index cd619acf..65128be4 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQBinderConstants.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQBinderConstants.java @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2018 the original author or authors. + * + * 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.springframework.cloud.stream.binder.rocketmq; /** diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java index a43b189c..d5109cda 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2018 the original author or authors. + * + * 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.springframework.cloud.stream.binder.rocketmq; import org.slf4j.Logger; diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageHeaderAccessor.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageHeaderAccessor.java index e746933b..4f0ca012 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageHeaderAccessor.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageHeaderAccessor.java @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2018 the original author or authors. + * + * 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.springframework.cloud.stream.binder.rocketmq; import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ACKNOWLEDGEMENT_KEY; diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/actuator/RocketMQBinderEndpoint.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/actuator/RocketMQBinderEndpoint.java index 5c4f9e67..1a7dc3b1 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/actuator/RocketMQBinderEndpoint.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/actuator/RocketMQBinderEndpoint.java @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2018 the original author or authors. + * + * 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.springframework.cloud.stream.binder.rocketmq.actuator; import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ENDPOINT_ID; diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/actuator/RocketMQBinderHealthIndicator.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/actuator/RocketMQBinderHealthIndicator.java index b517a6f7..12ed17eb 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/actuator/RocketMQBinderHealthIndicator.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/actuator/RocketMQBinderHealthIndicator.java @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2018 the original author or authors. + * + * 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.springframework.cloud.stream.binder.rocketmq.actuator; import org.springframework.beans.factory.annotation.Autowired; diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderAutoConfiguration.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderAutoConfiguration.java index dda8d439..95abaeeb 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderAutoConfiguration.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderAutoConfiguration.java @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2018 the original author or authors. + * + * 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.springframework.cloud.stream.binder.rocketmq.config; import org.apache.rocketmq.client.log.ClientLogger; diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderEndpointAutoConfiguration.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderEndpointAutoConfiguration.java index 7e40cd5b..f1fb5f1b 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderEndpointAutoConfiguration.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderEndpointAutoConfiguration.java @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2018 the original author or authors. + * + * 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.springframework.cloud.stream.binder.rocketmq.config; import org.springframework.boot.actuate.autoconfigure.endpoint.EndpointAutoConfiguration; diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/consuming/Acknowledgement.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/consuming/Acknowledgement.java index 1efe40e4..4b5d9e3f 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/consuming/Acknowledgement.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/consuming/Acknowledgement.java @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2018 the original author or authors. + * + * 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.springframework.cloud.stream.binder.rocketmq.consuming; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/consuming/ConsumersManager.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/consuming/ConsumersManager.java index 2dd425b0..12fde2ba 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/consuming/ConsumersManager.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/consuming/ConsumersManager.java @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2018 the original author or authors. + * + * 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.springframework.cloud.stream.binder.rocketmq.consuming; import java.util.AbstractMap; diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQInboundChannelAdapter.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQInboundChannelAdapter.java index 1831dd7b..7b70c067 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQInboundChannelAdapter.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQInboundChannelAdapter.java @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2018 the original author or authors. + * + * 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.springframework.cloud.stream.binder.rocketmq.integration; import java.util.ArrayList; diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQMessageHandler.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQMessageHandler.java index 1af26139..ece595e6 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQMessageHandler.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQMessageHandler.java @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2018 the original author or authors. + * + * 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.springframework.cloud.stream.binder.rocketmq.integration; import java.time.Instant; diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ConsumerGroupInstrumentation.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ConsumerGroupInstrumentation.java index 87f0c978..6e4386d0 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ConsumerGroupInstrumentation.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ConsumerGroupInstrumentation.java @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2018 the original author or authors. + * + * 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.springframework.cloud.stream.binder.rocketmq.metrics; import com.codahale.metrics.MetricRegistry; diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ConsumerInstrumentation.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ConsumerInstrumentation.java index ad72f4d8..787e4b83 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ConsumerInstrumentation.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ConsumerInstrumentation.java @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2018 the original author or authors. + * + * 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.springframework.cloud.stream.binder.rocketmq.metrics; import static com.codahale.metrics.MetricRegistry.name; diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/Instrumentation.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/Instrumentation.java index 08ba4dd1..d9997ce3 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/Instrumentation.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/Instrumentation.java @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2018 the original author or authors. + * + * 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.springframework.cloud.stream.binder.rocketmq.metrics; import java.util.concurrent.atomic.AtomicBoolean; diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/InstrumentationManager.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/InstrumentationManager.java index 811ba01d..c1c1cead 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/InstrumentationManager.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/InstrumentationManager.java @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2018 the original author or authors. + * + * 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.springframework.cloud.stream.binder.rocketmq.metrics; import java.util.HashMap; diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ProducerInstrumentation.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ProducerInstrumentation.java index 68c9c0ee..1ede7802 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ProducerInstrumentation.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ProducerInstrumentation.java @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2018 the original author or authors. + * + * 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.springframework.cloud.stream.binder.rocketmq.metrics; import static com.codahale.metrics.MetricRegistry.name; diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQBinderConfigurationProperties.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQBinderConfigurationProperties.java index 087d95c2..5c12a3b4 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQBinderConfigurationProperties.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQBinderConfigurationProperties.java @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2018 the original author or authors. + * + * 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.springframework.cloud.stream.binder.rocketmq.properties; import org.springframework.boot.context.properties.ConfigurationProperties; diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQBindingProperties.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQBindingProperties.java index bd5c6d13..094dafd1 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQBindingProperties.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQBindingProperties.java @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2018 the original author or authors. + * + * 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.springframework.cloud.stream.binder.rocketmq.properties; /** diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQConsumerProperties.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQConsumerProperties.java index 2aa12dbe..6343f153 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQConsumerProperties.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQConsumerProperties.java @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2018 the original author or authors. + * + * 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.springframework.cloud.stream.binder.rocketmq.properties; import org.apache.rocketmq.client.consumer.MQPushConsumer; @@ -12,68 +28,67 @@ import org.apache.rocketmq.common.protocol.heartbeat.MessageModel; */ public class RocketMQConsumerProperties { - /** - * using '||' to split tag - * {@link MQPushConsumer#subscribe(String, String)} - */ - private String tags; + /** + * using '||' to split tag {@link MQPushConsumer#subscribe(String, String)} + */ + private String tags; - /** + /** * {@link MQPushConsumer#subscribe(String, MessageSelector)} * {@link MessageSelector#bySql(String)} */ - private String sql; + private String sql; - /** - * {@link MessageModel#BROADCASTING} - */ - private Boolean broadcasting = false; + /** + * {@link MessageModel#BROADCASTING} + */ + private Boolean broadcasting = false; - /** - * if orderly is true, using {@link MessageListenerOrderly} - * else if orderly if false, using {@link MessageListenerConcurrently} - */ - private Boolean orderly = false; + /** + * if orderly is true, using {@link MessageListenerOrderly} else if orderly if false, + * using {@link MessageListenerConcurrently} + */ + private Boolean orderly = false; - private Boolean enabled = true; + private Boolean enabled = true; - public String getTags() { - return tags; - } + public String getTags() { + return tags; + } - public void setTags(String tags) { - this.tags = tags; - } + public void setTags(String tags) { + this.tags = tags; + } - public String getSql() { - return sql; - } + public String getSql() { + return sql; + } - public void setSql(String sql) { - this.sql = sql; - } + public void setSql(String sql) { + this.sql = sql; + } - public Boolean getOrderly() { - return orderly; - } + public Boolean getOrderly() { + return orderly; + } - public void setOrderly(Boolean orderly) { - this.orderly = orderly; - } + public void setOrderly(Boolean orderly) { + this.orderly = orderly; + } - public Boolean getEnabled() { - return enabled; - } + public Boolean getEnabled() { + return enabled; + } - public void setEnabled(Boolean enabled) { - this.enabled = enabled; - } + public void setEnabled(Boolean enabled) { + this.enabled = enabled; + } - public Boolean getBroadcasting() { - return broadcasting; - } + public Boolean getBroadcasting() { + return broadcasting; + } - public void setBroadcasting(Boolean broadcasting) { - this.broadcasting = broadcasting; - } + public void setBroadcasting(Boolean broadcasting) { + this.broadcasting = broadcasting; + } } diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQExtendedBindingProperties.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQExtendedBindingProperties.java index b4d507e3..b016f693 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQExtendedBindingProperties.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQExtendedBindingProperties.java @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2018 the original author or authors. + * + * 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.springframework.cloud.stream.binder.rocketmq.properties; import java.util.HashMap; diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQProducerProperties.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQProducerProperties.java index 828df00e..6afefd3b 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQProducerProperties.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQProducerProperties.java @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2018 the original author or authors. + * + * 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.springframework.cloud.stream.binder.rocketmq.properties; import org.apache.rocketmq.client.producer.DefaultMQProducer; diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/provisioning/RocketMQTopicProvisioner.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/provisioning/RocketMQTopicProvisioner.java index 210209cd..3871470f 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/provisioning/RocketMQTopicProvisioner.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/provisioning/RocketMQTopicProvisioner.java @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2018 the original author or authors. + * + * 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.springframework.cloud.stream.binder.rocketmq.provisioning; import org.apache.rocketmq.client.Validators; From c665e97fb170f4ad8eedbd8e3b134e9a9292f9e6 Mon Sep 17 00:00:00 2001 From: fangjian0423 Date: Fri, 30 Nov 2018 16:46:59 +0800 Subject: [PATCH 09/10] remove unuseful logging --- .../binder/rocketmq/provisioning/RocketMQTopicProvisioner.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/provisioning/RocketMQTopicProvisioner.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/provisioning/RocketMQTopicProvisioner.java index 3871470f..456ef5cc 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/provisioning/RocketMQTopicProvisioner.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/provisioning/RocketMQTopicProvisioner.java @@ -60,8 +60,7 @@ public class RocketMQTopicProvisioner implements Validators.checkTopic(topic); } catch (MQClientException e) { - logger.error("topic check error: " + topic, e); - throw new AssertionError(e); // Can't happen + throw new AssertionError(e); } } From 25aecfe1ad95828ec9377cf488e6884914480109 Mon Sep 17 00:00:00 2001 From: fangjian0423 Date: Fri, 30 Nov 2018 16:47:19 +0800 Subject: [PATCH 10/10] update docs --- .../src/main/asciidoc-zh/rocketmq.adoc | 248 +++++++++++++++++- .../src/main/asciidoc-zh/sentinel.adoc | 70 ++--- .../rocketmq-example/readme-zh.md | 53 ++-- 3 files changed, 313 insertions(+), 58 deletions(-) diff --git a/spring-cloud-alibaba-docs/src/main/asciidoc-zh/rocketmq.adoc b/spring-cloud-alibaba-docs/src/main/asciidoc-zh/rocketmq.adoc index 238f0b4b..5443b7dd 100644 --- a/spring-cloud-alibaba-docs/src/main/asciidoc-zh/rocketmq.adoc +++ b/spring-cloud-alibaba-docs/src/main/asciidoc-zh/rocketmq.adoc @@ -1 +1,247 @@ -== Spring Cloud Alibaba Rocket Binder +== Spring Cloud Alibaba RocketMQ Binder + +### RocketMQ 介绍 + +https://rocketmq.apache.org[RocketMQ] 是一款开源的分布式消息系统，基于高可用分布式集群技术，提供低延时的、高可靠的消息发布与订阅服务。同时，广泛应用于多个领域，包括异步通信解耦、企业解决方案、金融支付、电信、电子商务、快递物流、广告营销、社交、即时通信、移动应用、手游、视频、物联网、车联网等。 + +具有以下特点： + +* 能够保证严格的消息顺序 + +* 提供丰富的消息拉取模式 + +* 高效的订阅者水平扩展能力 + +* 实时的消息订阅机制 + +* 亿级消息堆积能力 + +### RocketMQ 基本使用 + +* 下载 RocketMQ + +下载 https://www.apache.org/dyn/closer.cgi?path=rocketmq/4.3.2/rocketmq-all-4.3.2-bin-release.zip[RocketMQ最新的二进制文件]，并解压 + +解压后的目录结构如下： + +``` +apache-rocketmq +├── LICENSE +├── NOTICE +├── README.md +├── benchmark +├── bin +├── conf +└── lib +``` + +* 启动 NameServer + +```bash +nohup sh bin/mqnamesrv & +tail -f ~/logs/rocketmqlogs/namesrv.log +``` + +* 启动 Broker + +```bash +nohup sh bin/mqbroker -n localhost:9876 & +tail -f ~/logs/rocketmqlogs/broker.log +``` + +* 发送、接收消息 + +发送消息： + +```bash +sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer +``` + +发送成功后显示：`SendResult [sendStatus=SEND_OK, msgId= ...` + +接收消息： + +```bash +sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer +``` + +接收成功后显示：`ConsumeMessageThread_%d Receive New Messages: [MessageExt...` + +* 关闭 Server + +```bash +sh bin/mqshutdown broker +sh bin/mqshutdown namesrv +``` + +### Spring Cloud Stream 介绍 + +Spring Cloud Stream 是一个用于构建基于消息的微服务应用框架。它基于 SpringBoot 来创建具有生产级别的单机 Spring 应用，并且使用 `Spring Integration` 与 Broker 进行连接。 + +Spring Cloud Stream 提供了消息中间件配置的统一抽象，推出了 publish-subscribe、consumer groups、partition 这些统一的概念。 + +Spring Cloud Stream 内部有两个概念：Binder 和 Binding。 + +* Binder: 跟外部消息中间件集成的组件，用来创建 Binding，各消息中间件都有自己的 Binder 实现。 + +比如 `Kafka` 的实现 `KafkaMessageChannelBinder`，`RabbitMQ` 的实现 `RabbitMessageChannelBinder` 以及 `RocketMQ` 的实现 `RocketMQMessageChannelBinder`。 + +* Binding: 包括 Input Binding 和 Output Binding。 + +Binding 在消息中间件与应用程序提供的 Provider 和 Consumer 之间提供了一个桥梁，实现了开发者只需使用应用程序的 Provider 或 Consumer 生产或消费数据即可，屏蔽了开发者与底层消息中间件的接触。 + +.Spring Cloud Stream +image::https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/images/SCSt-overview.png[] + +使用 Spring Cloud Stream 完成一段简单的消息发送和消息接收代码： + +```java +MessageChannel messageChannel = new DirectChannel(); + +// 消息订阅 +((SubscribableChannel) messageChannel).subscribe(new MessageHandler() { + @Override + public void handleMessage(Message message) throws MessagingException { + System.out.println("receive msg: " + message.getPayload()); + } +}); + +// 消息发送 +messageChannel.send(MessageBuilder.withPayload("simple msg").build()); +``` + +这段代码所有的消息类都是 `spring-messaging` 模块里提供的。屏蔽具体消息中间件的底层实现，如果想用更换消息中间件，在配置文件里配置相关消息中间件信息以及修改 binder 依赖即可。 + +**Spring Cloud Stream 底层也是基于这段代码去做了各种抽象。** + +### Spring Cloud Alibaba RocketMQ Binder 实现原理 + +.RocketMQ Binder处理流程 +image::https://cdn.nlark.com/lark/0/2018/png/64647/1543560843558-24525bf4-1d0e-4e10-be5f-bdde7127f6e6.png[] + + +RocketMQ Binder 的核心主要就是这3个类：`RocketMQMessageChannelBinder`，`RocketMQInboundChannelAdapter` 和 `RocketMQMessageHandler`。 + +`RocketMQMessageChannelBinder` 是个标准的 Binder 实现，其内部构建 `RocketMQInboundChannelAdapter` 和 `RocketMQMessageHandler`。 + +`RocketMQMessageHandler` 用于 RocketMQ `Producer` 的启动以及消息的发送，其内部会根据 `spring-messaging` 模块内 `org.springframework.messaging.Message` 消息类，去创建 RocketMQ 的消息类 `org.apache.rocketmq.common.message.Message`。 + +在构造 `org.apache.rocketmq.common.message.Message` 的过程中会根据 `org.springframework.messaging.Message` 的 Header 构造成 `RocketMQMessageHeaderAccessor`。然后再根据 `RocketMQMessageHeaderAccessor` 中的一些属性，比如 tags、keys、flag等属性设置到 RocketMQ 的消息类 `org.apache.rocketmq.common.message.Message` 中。 + +`RocketMQInboundChannelAdapter` 用于 RocketMQ `Consumer` 的启动以及消息的接收。其内部还支持 https://github.com/spring-projects/spring-retry[spring-retry] 的使用。 + +在消费消息的时候可以从 Header 中获取 `Acknowledgement` 并进行一些设置。 + +比如使用 `MessageListenerConcurrently` 进行异步消费的时候，可以设置延迟消费： + +```java +@StreamListener("input") +public void receive(Message message) { + RocketMQMessageHeaderAccessor headerAccessor = new RocketMQMessageHeaderAccessor(message); + Acknowledgement acknowledgement = headerAccessor.getAcknowledgement(message); + acknowledgement.setConsumeConcurrentlyStatus(ConsumeConcurrentlyStatus.RECONSUME_LATER); + acknowledgement.setConsumeConcurrentlyDelayLevel(1); +} +``` + +比如使用 `MessageListenerOrderly` 进行顺序消费的时候，可以设置延迟消费： + +```java +@StreamListener("input") +public void receive(Message message) { + RocketMQMessageHeaderAccessor headerAccessor = new RocketMQMessageHeaderAccessor(message); + Acknowledgement acknowledgement = headerAccessor.getAcknowledgement(message); + acknowledgement.setConsumeOrderlyStatus(ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT); + acknowledgement.setConsumeOrderlySuspendCurrentQueueTimeMill(5000); +} +``` + +Provider端支持的配置： + +:frame: topbot +[width="60%",options="header"] +|==== +^|配置项 ^|含义 ^| 默认值 +|`spring.cloud.stream.rocketmq.bindings.your-output-binding.producer.enabled`|是否启用producer|true +|`spring.cloud.stream.rocketmq.bindings.your-output-binding.producer.max-message-size`|消息发送的最大字节数|0(大于0才会生效，RocketMQ 默认值为4M = 1024 * 1024 * 4) +|==== + +Consumer端支持的配置： + +:frame: topbot +[width="60%",options="header"] +|==== +^|配置项 ^|含义| 默认值 +|`spring.cloud.stream.rocketmq.bindings.your-input-binding.consumer.enabled`|是否启用consumer|true +|`spring.cloud.stream.rocketmq.bindings.your-input-binding.consumer.tags`|Consumer订阅只有包括这些tags的topic消息。多个标签之间使用 "\|\|" 分割| +|`spring.cloud.stream.rocketmq.bindings.your-input-binding.consumer.sql`|Consumer订阅满足sql要求的topic消息| +|`spring.cloud.stream.rocketmq.bindings.your-input-binding.consumer.broadcasting`|Consumer是否是广播模式|false +|`spring.cloud.stream.rocketmq.bindings.your-input-binding.consumer.orderly`|顺序消费 or 异步消费|false +|==== + +### Endpoint支持 + +在使用Endpoint特性之前需要在 Maven 中添加 `spring-boot-starter-actuator` 依赖，并在配置中允许 Endpoints 的访问。 + +* Spring Boot 1.x 中添加配置 `management.security.enabled=false`。暴露的 endpoint 路径为 `/rocketmq_binder` +* Spring Boot 2.x 中添加配置 `management.endpoints.web.exposure.include=*`。暴露的 endpoint 路径为 `/actuator/rocketmq-binder` + +Endpoint 会统计消息最后一次发送的数据，消息发送成功或失败的次数，消息消费成功或失败的次数等数据。 + +```json +{ + "runtime": { + "lastSend.timestamp": 1542786623915 + }, + "metrics": { + "scs-rocketmq.consumer.test-topic.totalConsumed": { + "count": 11 + }, + "scs-rocketmq.consumer.test-topic.totalConsumedFailures": { + "count": 0 + }, + "scs-rocketmq.producer.test-topic.totalSentFailures": { + "count": 0 + }, + "scs-rocketmq.consumer.test-topic.consumedPerSecond": { + "count": 11, + "fifteenMinuteRate": 0.012163847780107841, + "fiveMinuteRate": 0.03614605351360527, + "meanRate": 0.3493213353657594, + "oneMinuteRate": 0.17099243039490175 + }, + "scs-rocketmq.producer.test-topic.totalSent": { + "count": 5 + }, + "scs-rocketmq.producer.test-topic.sentPerSecond": { + "count": 5, + "fifteenMinuteRate": 0.005540151995103271, + "fiveMinuteRate": 0.01652854617838251, + "meanRate": 0.10697493212602836, + "oneMinuteRate": 0.07995558537067671 + }, + "scs-rocketmq.producer.test-topic.sentFailuresPerSecond": { + "count": 0, + "fifteenMinuteRate": 0.0, + "fiveMinuteRate": 0.0, + "meanRate": 0.0, + "oneMinuteRate": 0.0 + }, + "scs-rocketmq.consumer.test-topic.consumedFailuresPerSecond": { + "count": 0, + "fifteenMinuteRate": 0.0, + "fiveMinuteRate": 0.0, + "meanRate": 0.0, + "oneMinuteRate": 0.0 + } + } +} +``` + +注意：要想查看统计数据需要在pom里加上 https://mvnrepository.com/artifact/io.dropwizard.metrics/metrics-core[metrics-core依赖]。如若不加，endpoint 将会显示 warning 信息而不会显示统计信息： + +```json +{ + "warning": "please add metrics-core dependency, we use it for metrics" +} +``` \ No newline at end of file diff --git a/spring-cloud-alibaba-docs/src/main/asciidoc-zh/sentinel.adoc b/spring-cloud-alibaba-docs/src/main/asciidoc-zh/sentinel.adoc index e4c9817f..e8c575ac 100644 --- a/spring-cloud-alibaba-docs/src/main/asciidoc-zh/sentinel.adoc +++ b/spring-cloud-alibaba-docs/src/main/asciidoc-zh/sentinel.adoc @@ -1,20 +1,20 @@ == Spring Cloud Alibaba Sentinel -### Sentinel介绍 +### Sentinel 介绍 -随着微服务的流行，服务和服务之间的稳定性变得越来越重要。Sentinel 以流量为切入点，从流量控制、熔断降级、系统负载保护等多个维度保护服务的稳定性。 +随着微服务的流行，服务和服务之间的稳定性变得越来越重要。 https://github.com/alibaba/Sentinel[Sentinel] 以流量为切入点，从流量控制、熔断降级、系统负载保护等多个维度保护服务的稳定性。 -Sentinel 具有以下特征: +https://github.com/alibaba/Sentinel[Sentinel] 具有以下特征: -* *丰富的应用场景*：Sentinel 承接了阿里巴巴近 10 年的双十一大促流量的核心场景，例如秒杀（即突发流量控制在系统容量可以承受的范围）、消息削峰填谷、实时熔断下游不可用应用等。 -* *完备的实时监控*：Sentinel 同时提供实时的监控功能。您可以在控制台中看到接入应用的单台机器秒级数据，甚至 500 台以下规模的集群的汇总运行情况。 -* *广泛的开源生态*：Sentinel 提供开箱即用的与其它开源框架/库的整合模块，例如与 Spring Cloud、Dubbo、gRPC 的整合。您只需要引入相应的依赖并进行简单的配置即可快速地接入 Sentinel。 -* *完善的 SPI 扩展点*：Sentinel 提供简单易用、完善的 SPI 扩展点。您可以通过实现扩展点，快速的定制逻辑。例如定制规则管理、适配数据源等。 +* *丰富的应用场景*： Sentinel 承接了阿里巴巴近 10 年的双十一大促流量的核心场景，例如秒杀（即突发流量控制在系统容量可以承受的范围）、消息削峰填谷、实时熔断下游不可用应用等。 +* *完备的实时监控*： Sentinel 同时提供实时的监控功能。您可以在控制台中看到接入应用的单台机器秒级数据，甚至 500 台以下规模的集群的汇总运行情况。 +* *广泛的开源生态*： Sentinel 提供开箱即用的与其它开源框架/库的整合模块，例如与 Spring Cloud、Dubbo、gRPC 的整合。您只需要引入相应的依赖并进行简单的配置即可快速地接入 Sentinel。 +* *完善的 SPI 扩展点*： Sentinel 提供简单易用、完善的 SPI 扩展点。您可以通过实现扩展点，快速的定制逻辑。例如定制规则管理、适配数据源等。 -### 如何使用Sentinel +### 如何使用 Sentinel -如果要在您的项目中引入Sentinel，使用group ID为 `org.springframework.cloud` 和artifact ID为 `spring-cloud-starter-alibaba-sentinel` 的starter。 +如果要在您的项目中引入 Sentinel，使用 group ID 为 `org.springframework.cloud` 和 artifact ID 为 `spring-cloud-starter-alibaba-sentinel` 的 starter。 ```xml @@ -23,7 +23,7 @@ Sentinel 具有以下特征: ``` -下面这个例子就是一个最简单的使用Sentinel的例子: +下面这个例子就是一个最简单的使用 Sentinel 的例子: ```java @SpringBootApplication @@ -47,9 +47,9 @@ public class TestController { } ``` -@SentinelResource注解用来标识资源是否被限流、降级。上述例子上该注解的属性'hello'表示资源名。 +@SentinelResource 注解用来标识资源是否被限流、降级。上述例子上该注解的属性 'hello' 表示资源名。 -@SentinelResource还提供了其它额外的属性如 `blockHandler`，`blockHandlerClass`，`fallback` 用于表示限流或降级的操作，更多内容可以参考 https://github.com/alibaba/Sentinel/wiki/%E6%B3%A8%E8%A7%A3%E6%94%AF%E6%8C%81[Sentinel注解支持]。 +@SentinelResource 还提供了其它额外的属性如 `blockHandler`，`blockHandlerClass`，`fallback` 用于表示限流或降级的操作，更多内容可以参考 https://github.com/alibaba/Sentinel/wiki/%E6%B3%A8%E8%A7%A3%E6%94%AF%E6%8C%81[Sentinel注解支持]。 ##### Sentinel 控制台 @@ -74,7 +74,7 @@ image::https://github.com/alibaba/Sentinel/wiki/image/dashboard.png[] ###### 启动控制台 -Sentinel 控制台是一个标准的SpringBoot应用，以SpringBoot的方式运行jar包即可。 +Sentinel 控制台是一个标准的 SpringBoot 应用，以 SpringBoot 的方式运行 jar 包即可。 ```shell java -Dserver.port=8080 -Dcsp.sentinel.dashboard.server=localhost:8080 -Dproject.name=sentinel-dashboard -jar sentinel-dashboard.jar @@ -94,17 +94,17 @@ spring: dashboard: localhost:8080 ---- -这里的 `spring.cloud.sentinel.transport.port` 端口配置会在应用对应的机器上启动一个Http Server，该Serve会与Sentinel控制台做交互。比如Sentinel控制台添加了1个限流规则，会把规则数据push给这个Http Server接受，Http Server再将规则注册到Sentinel中。 +这里的 `spring.cloud.sentinel.transport.port` 端口配置会在应用对应的机器上启动一个 Http Server，该 Server 会与 Sentinel 控制台做交互。比如 Sentinel 控制台添加了1个限流规则，会把规则数据 push 给这个 Http Server 接收，Http Server 再将规则注册到 Sentinel 中。 -更多Sentinel控制台的使用及问题参考： https://github.com/alibaba/Sentinel/wiki/%E6%8E%A7%E5%88%B6%E5%8F%B0[Sentinel控制台] +更多 Sentinel 控制台的使用及问题参考： https://github.com/alibaba/Sentinel/wiki/%E6%8E%A7%E5%88%B6%E5%8F%B0[Sentinel控制台] -### Feign支持 +### Feign 支持 DOING -### RestTemplate支持 +### RestTemplate 支持 -Spring Cloud Alibaba Sentinel支持对 `RestTemplate` 的服务调用使用Sentinel进行保护，在构造 `RestTemplate` bean的时候需要加上 `@SentinelProtect` 注解。 +Spring Cloud Alibaba Sentinel 支持对 `RestTemplate` 的服务调用使用 Sentinel 进行保护，在构造 `RestTemplate` bean的时候需要加上 `@SentinelProtect` 注解。 ```java @Bean @@ -122,12 +122,12 @@ public RestTemplate restTemplate() { * `schema://host:port`：协议、主机和端口 -NOTE: 以 `https://www.taobao.com/test` 这个url为例。对应的资源名有两种粒度，分别是 `https://www.taobao.com:80` 以及 `https://www.taobao.com:80/test` +NOTE: 以 `https://www.taobao.com/test` 这个 url 为例。对应的资源名有两种粒度，分别是 `https://www.taobao.com:80` 以及 `https://www.taobao.com:80/test` ### 动态数据源支持 -*在版本 0.2.0.RELEASE 或 0.1.0.RELEASE 之前*，要在Spring Cloud Alibaba Sentinel下使用动态数据源，需要3个步骤： +*在版本 0.2.0.RELEASE 或 0.1.0.RELEASE 之前*，要在 Spring Cloud Alibaba Sentinel 下使用动态数据源，需要3个步骤： * 配置文件中定义数据源信息。比如使用文件： @@ -141,7 +141,7 @@ spring.cloud.sentinel.datasource.converter=flowConverter spring.cloud.sentinel.datasource.file=/Users/you/yourrule.json ---- -* 创建一个Converter类实现 `com.alibaba.csp.sentinel.datasource.Converter` 接口，并且需要在 `ApplicationContext` 中有该类的一个Bean +* 创建一个 Converter 类实现 `com.alibaba.csp.sentinel.datasource.Converter` 接口，并且需要在 `ApplicationContext` 中有该类的一个 Bean ```java @Component("flowConverter") @@ -154,7 +154,7 @@ public class JsonFlowRuleListParser implements Converter> } ``` -这个Converter的bean name需要跟 `application.properties` 配置文件中的converter配置一致 +这个 Converter 的 bean name 需要跟 `application.properties` 配置文件中的 converter 配置一致 * 在任意一个 Spring Bean 中定义一个被 `@SentinelDataSource` 注解修饰的 `ReadableDataSource` 属性 @@ -163,7 +163,7 @@ public class JsonFlowRuleListParser implements Converter> private ReadableDataSource dataSource; ``` -`@SentinelDataSource` 注解的value属性表示数据源在 `application.properties` 配置文件中前缀。 该在例子中，前缀为 `spring.cloud.sentinel.datasource` 。 +`@SentinelDataSource` 注解的 value 属性表示数据源在 `application.properties` 配置文件中前缀。 该在例子中，前缀为 `spring.cloud.sentinel.datasource` 。 如果 `ApplicationContext` 中存在超过1个 `ReadableDataSource` bean，那么不会加载这些 `ReadableDataSource` 中的任意一个。 只有在 `ApplicationContext` 存在一个 `ReadableDataSource` 的情况下才会生效。 @@ -173,7 +173,7 @@ private ReadableDataSource dataSource; [Sentinel Starter] load 3 flow rules ``` -*在版本 0.2.0.RELEASE 或 0.1.0.RELEASE 之后*，要在Spring Cloud Alibaba Sentinel下使用动态数据源，只需要1个步骤： +*在版本 0.2.0.RELEASE 或 0.1.0.RELEASE 之后*，要在 Spring Cloud Alibaba Sentinel 下使用动态数据源，只需要1个步骤： * 直接在 `application.properties` 配置文件中配置数据源信息即可 @@ -196,23 +196,23 @@ spring.cloud.sentinel.datasource.ds4.apollo.default-flow-rule-value = test ``` -这样配置方式参考了Spring Cloud Stream Binder的配置，内部使用了 `TreeMap` 进行存储，comparator为 `String.CASE_INSENSITIVE_ORDER` 。 +这样配置方式参考了 Spring Cloud Stream Binder 的配置，内部使用了 `TreeMap` 进行存储，comparator 为 `String.CASE_INSENSITIVE_ORDER` 。 NOTE: d1, ds2, ds3, ds4 是 `ReadableDataSource` 的名字，可随意编写。后面的 `file` ，`zk` ，`nacos` , `apollo` 就是对应具体的数据源。 它们后面的配置就是这些数据源各自的配置。 每种数据源都有两个共同的配置项： `data-type` 和 `converter-class` 。 -`data-type` 配置项表示 `Converter`，Spring Cloud Alibaba Sentinel默认提供两种内置的值，分别是 `json` 和 `xml` (不填默认是json)。 如果不想使用内置的 `json` 或 `xml` 这两种 `Converter`，可以填写 `custom` 表示自定义 `Converter`，然后再配置 `converter-class` 配置项，该配置项需要写类的全路径名。 +`data-type` 配置项表示 `Converter`，Spring Cloud Alibaba Sentinel 默认提供两种内置的值，分别是 `json` 和 `xml` (不填默认是json)。 如果不想使用内置的 `json` 或 `xml` 这两种 `Converter`，可以填写 `custom` 表示自定义 `Converter`，然后再配置 `converter-class` 配置项，该配置项需要写类的全路径名。 -这两种内置的 `Converter` 只支持解析 json数组 或 xml数组。内部解析的时候会自动判断每个json对象或xml对象属于哪4种Sentinel规则(`FlowRule`，`DegradeRule`，`SystemRule`，`AuthorityRule`)。 +这两种内置的 `Converter` 只支持解析 json 数组 或 xml 数组。内部解析的时候会自动判断每个 json 对象或xml对象属于哪4种 Sentinel 规则(`FlowRule`，`DegradeRule`，`SystemRule`，`AuthorityRule`)。 比如10个规则数组里解析出5个限流规则和5个降级规则。 这种情况下该数据源不会注册，日志里页会进行警告。 如果10个规则里有9个限流规则，1个解析报错了。这种情况下日志会警告有个规则格式错误，另外9个限流规则会注册上去。 -这里json或xml解析用的是 `jackson`。`ObjectMapper` 或 `XmlMapper` 使用默认的配置，遇到不认识的字段会解析报错。 因为不这样做的话限流 json 也会解析成 系统规则(系统规则所有的配置都有默认值)，所以需要这样严格解析。 +这里 json 或 xml 解析用的是 `jackson`。`ObjectMapper` 或 `XmlMapper` 使用默认的配置，遇到不认识的字段会解析报错。 因为不这样做的话限流 json 也会解析成 系统规则(系统规则所有的配置都有默认值)，所以需要这样严格解析。 -当然还有一种情况是json对象或xml对象可能会匹配上所有4种规则(比如json对象里只配了 `resource` 字段，那么会匹配上4种规则)，这种情况下日志里会警告，并且过滤掉这个对象。 +当然还有一种情况是 json 对象或 xml 对象可能会匹配上所有4种规则(比如json对象里只配了 `resource` 字段，那么会匹配上4种规则)，这种情况下日志里会警告，并且过滤掉这个对象。 用户使用这种配置的时候只需要填写正确的json或xml就行，有任何不合理的信息都会在日志里打印出来。 @@ -223,16 +223,16 @@ NOTE: d1, ds2, ds3, ds4 是 `ReadableDataSource` 的名字，可随意编写。 [Sentinel Starter] DataSource ds2-sentinel-nacos-datasource load 2 FlowRule ``` -NOTE: 默认情况下，xml格式是不支持的。需要添加 `jackson-dataformat-xml` 依赖后才会自动生效。 +NOTE: 默认情况下，xml 格式是不支持的。需要添加 `jackson-dataformat-xml` 依赖后才会自动生效。 -关于Sentinel动态数据源的实现原理，参考： https://github.com/alibaba/Sentinel/wiki/%E5%8A%A8%E6%80%81%E8%A7%84%E5%88%99%E6%89%A9%E5%B1%95[动态规则扩展] +关于 Sentinel 动态数据源的实现原理，参考： https://github.com/alibaba/Sentinel/wiki/%E5%8A%A8%E6%80%81%E8%A7%84%E5%88%99%E6%89%A9%E5%B1%95[动态规则扩展] -### Endpoint支持 +### Endpoint 支持 -在使用Endpoint特性之前需要在 Maven 中添加 `spring-boot-starter-actuator` 依赖，并在配置中允许 Endpoints 的访问。 +在使用 Endpoint 特性之前需要在 Maven 中添加 `spring-boot-starter-actuator` 依赖，并在配置中允许 Endpoints 的访问。 -* Spring Boot 1.x 中添加配置 `management.security.enabled=false`。暴露的endpoint路径为 `/sentinel` -* Spring Boot 2.x 中添加配置 `management.endpoints.web.exposure.include=*`。暴露的endpoint路径为 `/actuator/sentinel` +* Spring Boot 1.x 中添加配置 `management.security.enabled=false`。暴露的 endpoint 路径为 `/sentinel` +* Spring Boot 2.x 中添加配置 `management.endpoints.web.exposure.include=*`。暴露的 endpoint 路径为 `/actuator/sentinel` ### More diff --git a/spring-cloud-alibaba-examples/rocketmq-example/readme-zh.md b/spring-cloud-alibaba-examples/rocketmq-example/readme-zh.md index bd724a53..7bb2472f 100644 --- a/spring-cloud-alibaba-examples/rocketmq-example/readme-zh.md +++ b/spring-cloud-alibaba-examples/rocketmq-example/readme-zh.md @@ -6,17 +6,25 @@ [RocketMQ](https://rocketmq.apache.org/) 是一款开源的分布式消息系统，基于高可用分布式集群技术，提供低延时的、高可靠的消息发布与订阅服务。 -在说明RocketMQ的示例之前，我们先了解一下 Spring Cloud Stream 中的Binder和Binding概念。 +在说明 RocketMQ 的示例之前，我们先了解一下 Spring Cloud Stream。 -Binder: 跟外部消息中间件集成的组件，用来创建Binding，各消息中间件都有自己的Binder实现。 +这是官方对 Spring Cloud Stream 的一段介绍： -比如 `Kafka` 的实现 `KafkaMessageChannelBinder` ，`RabbitMQ` 的实现 `RabbitMessageChannelBinder` 以及 `RocketMQ` 的实现 `RocketMQMessageChannelBinder` 。 +Spring Cloud Stream 是一个用于构建基于消息的微服务应用框架。它基于 SpringBoot 来创建具有生产级别的单机 Spring 应用，并且使用 `Spring Integration` 与 Broker 进行连接。 + +Spring Cloud Stream 提供了消息中间件配置的统一抽象，推出了 publish-subscribe、consumer groups、partition 这些统一的概念。 -Binding: 包括Input Binding和Output Binding。 +Spring Cloud Stream 内部有两个概念：Binder 和 Binding。 -Binding在消息中间件与应用程序提供的Provider和Consumer之间提供了一个桥梁，实现了开发者只需使用应用程序的Provider或Consumer生产或消费数据即可，屏蔽了开发者与底层消息中间件的接触。 +* Binder: 跟外部消息中间件集成的组件，用来创建 Binding，各消息中间件都有自己的 Binder 实现。 -下图是Spring Cloud Stream的架构设计。 +比如 `Kafka` 的实现 `KafkaMessageChannelBinder`，`RabbitMQ` 的实现 `RabbitMessageChannelBinder` 以及 `RocketMQ` 的实现 `RocketMQMessageChannelBinder`。 + +* Binding: 包括 Input Binding 和 Output Binding。 + +Binding 在消息中间件与应用程序提供的 Provider 和 Consumer 之间提供了一个桥梁，实现了开发者只需使用应用程序的 Provider 或 Consumer 生产或消费数据即可，屏蔽了开发者与底层消息中间件的接触。 + +下图是 Spring Cloud Stream 的架构设计。 ![](https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/images/SCSt-overview.png) @@ -37,7 +45,7 @@ Binding在消息中间件与应用程序提供的Provider和Consumer之间提供 ``` -2. 配置Input和Output的Binding信息并配合`@EnableBinding`注解使其生效 +2. 配置 Input 和 Output 的 Binding 信息并配合 `@EnableBinding` 注解使其生效 ```java @SpringBootApplication @@ -49,7 +57,7 @@ public class RocketMQApplication { } ``` -配置Binding信息： +配置 Binding 信息： ```properties # 配置rocketmq的nameserver地址 spring.cloud.stream.rocketmq.binder.namesrv-addr=127.0.0.1:9876 @@ -67,7 +75,7 @@ spring.cloud.stream.bindings.input.group=test-group ### 下载并启动 RocketMQ -在接入RocketMQ Binder之前，首先需要启动RocketMQ的Name Server和Broker。 +在接入 RocketMQ Binder 之前，首先需要启动 RocketMQ 的 Name Server 和 Broker。 1. 下载[RocketMQ最新的二进制文件](https://www.apache.org/dyn/closer.cgi?path=rocketmq/4.3.2/rocketmq-all-4.3.2-bin-release.zip)，并解压 @@ -77,13 +85,13 @@ spring.cloud.stream.bindings.input.group=test-group sh bin/mqnamesrv ``` -3. 启动Broker +3. 启动 Broker ```bash sh bin/mqbroker -n localhost:9876 ``` -4. 创建Topic: test-topic +4. 创建 Topic: test-topic ```bash sh bin/mqadmin updateTopic -n localhost:9876 -c DefaultCluster -t test-topic @@ -100,18 +108,19 @@ server.port=28081 2. 启动应用，支持 IDE 直接启动和编译打包后启动。 - 1. IDE直接启动：找到主类 `RocketMQApplication`，执行 main 方法启动应用。 - 2. 打包编译后启动：首先执行 `mvn clean package` 将工程编译打包，然后执行 `java -jar rocketmq-example.jar`启动应用。 + 1. IDE 直接启动：找到主类 `RocketMQApplication`，执行 main 方法启动应用。 + 2. 打包编译后启动：首先执行 `mvn clean package` 将工程编译打包，然后执行 `java -jar rocketmq-example.jar` 启动应用。 ### 消息处理 -使用name为output对应的binding发送消息到test-topic这个topic。 +使用 name 为 output 对应的 binding 发送消息到 test-topic 这个 topic。 -使用2个input binding订阅数据。 +使用2个 input binding 订阅数据。 -input1: 订阅topic为test-topic的消息，顺序消费所有消息(顺序消费的前提是所有消息都在一个MessageQueue中) -input2: 订阅topic为test-topic的消息，异步消费tags为tagStr的消息，Consumer端线程池个数为20 +* input1: 订阅 topic 为 test-topic 的消息，顺序消费所有消息(顺序消费的前提是所有消息都在一个 MessageQueue 中) + +* input2: 订阅 topic 为 test-topic 的消息，异步消费 tags 为 tagStr 的消息，Consumer 端线程池个数为20 配置信息如下： @@ -137,7 +146,7 @@ spring.cloud.stream.bindings.input2.consumer.concurrency=20 #### 消息发送 -使用MessageChannel进行消息发送： +使用 MessageChannel 进行消息发送： ```java public class ProducerRunner implements CommandLineRunner { @@ -153,7 +162,7 @@ public class ProducerRunner implements CommandLineRunner { } ``` -或者使用RocketMQ原生的API进行消息发送: +或者使用 RocketMQ 原生的 API 进行消息发送: ```java public class RocketMQProducer { @@ -168,7 +177,7 @@ public class RocketMQProducer { #### 消息接收 -使用`@StreamListener`注解接收消息： +使用 `@StreamListener` 注解接收消息： ```java @Service @@ -249,11 +258,11 @@ Spring Boot 1.x 可以通过访问 http://127.0.0.1:18083/rocketmq_binder 来查 } ``` -注意：要想查看统计数据需要在pom里加上 [metrics-core依赖](https://mvnrepository.com/artifact/io.dropwizard.metrics/metrics-core) 。如若不加，endpoint将会显示warning信息而不会显示统计信息： +注意：要想查看统计数据需要在pom里加上 [metrics-core依赖](https://mvnrepository.com/artifact/io.dropwizard.metrics/metrics-core)。如若不加，endpoint 将会显示 warning 信息而不会显示统计信息： ```json { - "warning": "please add metrics-core dependency, we use it to metrics" + "warning": "please add metrics-core dependency, we use it for metrics" } ```