update 增加 spring-cloud-bus redis实现 workflow模块默认使用redis做事件中心 不用强制使用mq了(不保证稳定 需进行长期测试)
This commit is contained in:
疯狂的狮子Li
@@ -17,10 +17,10 @@
|
||||
|
||||
<dependencies>
|
||||
<!-- rabbitmq -->
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-starter-bus-amqp</artifactId>
|
||||
</dependency>
|
||||
<!-- <dependency>-->
|
||||
<!-- <groupId>org.springframework.cloud</groupId>-->
|
||||
<!-- <artifactId>spring-cloud-starter-bus-amqp</artifactId>-->
|
||||
<!-- </dependency>-->
|
||||
<!-- kafka -->
|
||||
<!-- <dependency>-->
|
||||
<!-- <groupId>org.springframework.cloud</groupId>-->
|
||||
@@ -31,5 +31,25 @@
|
||||
<!-- <groupId>com.alibaba.cloud</groupId>-->
|
||||
<!-- <artifactId>spring-cloud-starter-bus-rocketmq</artifactId>-->
|
||||
<!-- </dependency>-->
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-bus</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-health</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.dromara</groupId>
|
||||
<artifactId>ruoyi-common-redis</artifactId>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</project>
|
||||
|
||||
+41
@@ -0,0 +1,41 @@
|
||||
package org.dromara.common.bus.config;
|
||||
|
||||
import org.springframework.context.annotation.Condition;
|
||||
import org.springframework.context.annotation.ConditionContext;
|
||||
import org.springframework.core.io.Resource;
|
||||
import org.springframework.core.type.AnnotatedTypeMetadata;
|
||||
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
|
||||
import org.springframework.core.io.support.PropertiesLoaderUtils;
|
||||
import org.springframework.core.io.support.ResourcePatternResolver;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
import java.util.Properties;
|
||||
|
||||
/**
|
||||
* Matches when no Spring Cloud Stream binder is provided by external dependencies.
|
||||
*
|
||||
* @author dromara
|
||||
*/
|
||||
public class NoExternalStreamBinderCondition implements Condition {
|
||||
|
||||
private static final String BINDER_RESOURCE = "classpath*:META-INF/spring.binders";
|
||||
|
||||
@Override
|
||||
public boolean matches(ConditionContext context, AnnotatedTypeMetadata metadata) {
|
||||
try {
|
||||
ResourcePatternResolver resolver = context.getResourceLoader() instanceof ResourcePatternResolver resourcePatternResolver
|
||||
? resourcePatternResolver : new PathMatchingResourcePatternResolver(context.getResourceLoader());
|
||||
Resource[] resources = resolver.getResources(BINDER_RESOURCE);
|
||||
for (Resource resource : resources) {
|
||||
Properties properties = PropertiesLoaderUtils.loadProperties(resource);
|
||||
if (properties.keySet().stream().anyMatch(key -> StringUtils.hasText(String.valueOf(key)))) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
return true;
|
||||
} catch (Exception e) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
+45
@@ -0,0 +1,45 @@
|
||||
package org.dromara.common.bus.config;
|
||||
|
||||
import org.dromara.common.bus.redis.RedisMessageChannelBinder;
|
||||
import org.dromara.common.bus.redis.RedisBusHealthIndicator;
|
||||
import org.redisson.api.RedissonClient;
|
||||
import org.springframework.boot.autoconfigure.AutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.boot.health.contributor.HealthIndicator;
|
||||
import org.springframework.cloud.stream.binder.Binder;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Conditional;
|
||||
|
||||
/**
|
||||
* Redis stream binder for Spring Cloud Bus.
|
||||
*
|
||||
* @author dromara
|
||||
*/
|
||||
@AutoConfiguration
|
||||
@ConditionalOnClass({RedissonClient.class, Binder.class})
|
||||
@Conditional(NoExternalStreamBinderCondition.class)
|
||||
@EnableConfigurationProperties(RedisBusProperties.class)
|
||||
public class RedisBusBinderConfiguration {
|
||||
|
||||
/**
|
||||
* Use Redis as the default Spring Cloud Stream binder when no MQ binder exists.
|
||||
*/
|
||||
@Bean("redis")
|
||||
@ConditionalOnMissingBean(Binder.class)
|
||||
public RedisMessageChannelBinder redisMessageChannelBinder(RedissonClient redissonClient, RedisBusProperties properties) {
|
||||
return new RedisMessageChannelBinder(redissonClient, properties);
|
||||
}
|
||||
|
||||
/**
|
||||
* Redis bus health indicator.
|
||||
*/
|
||||
@Bean
|
||||
@ConditionalOnClass(HealthIndicator.class)
|
||||
@ConditionalOnMissingBean(name = "redisBusHealthIndicator")
|
||||
public RedisBusHealthIndicator redisBusHealthIndicator(RedissonClient redissonClient) {
|
||||
return new RedisBusHealthIndicator(redissonClient);
|
||||
}
|
||||
|
||||
}
|
||||
+106
@@ -0,0 +1,106 @@
|
||||
package org.dromara.common.bus.config;
|
||||
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
|
||||
import java.time.Duration;
|
||||
|
||||
/**
|
||||
* Redis bus binder properties.
|
||||
*
|
||||
* @author dromara
|
||||
*/
|
||||
@ConfigurationProperties(prefix = "spring.cloud.bus.redis")
|
||||
public class RedisBusProperties {
|
||||
|
||||
/**
|
||||
* Pending messages idle longer than this value can be claimed by another consumer.
|
||||
*/
|
||||
private Duration claimIdleTimeout = Duration.ofSeconds(30);
|
||||
|
||||
/**
|
||||
* Maximum pending messages to claim in one poll.
|
||||
*/
|
||||
private int claimCount = 10;
|
||||
|
||||
/**
|
||||
* Maximum new messages to read in one poll.
|
||||
*/
|
||||
private int readCount = 10;
|
||||
|
||||
/**
|
||||
* Blocking read timeout.
|
||||
*/
|
||||
private Duration readTimeout = Duration.ofSeconds(2);
|
||||
|
||||
/**
|
||||
* Maximum delivery attempts before sending to DLQ.
|
||||
*/
|
||||
private int maxAttempts = 3;
|
||||
|
||||
/**
|
||||
* Whether to write exhausted messages to a dead-letter stream.
|
||||
*/
|
||||
private boolean dlqEnabled = true;
|
||||
|
||||
/**
|
||||
* Maximum Redis Stream length. Redis applies this approximately.
|
||||
*/
|
||||
private int streamMaxLen = 10000;
|
||||
|
||||
public Duration getClaimIdleTimeout() {
|
||||
return claimIdleTimeout;
|
||||
}
|
||||
|
||||
public void setClaimIdleTimeout(Duration claimIdleTimeout) {
|
||||
this.claimIdleTimeout = claimIdleTimeout;
|
||||
}
|
||||
|
||||
public int getClaimCount() {
|
||||
return claimCount;
|
||||
}
|
||||
|
||||
public void setClaimCount(int claimCount) {
|
||||
this.claimCount = claimCount;
|
||||
}
|
||||
|
||||
public int getReadCount() {
|
||||
return readCount;
|
||||
}
|
||||
|
||||
public void setReadCount(int readCount) {
|
||||
this.readCount = readCount;
|
||||
}
|
||||
|
||||
public Duration getReadTimeout() {
|
||||
return readTimeout;
|
||||
}
|
||||
|
||||
public void setReadTimeout(Duration readTimeout) {
|
||||
this.readTimeout = readTimeout;
|
||||
}
|
||||
|
||||
public int getMaxAttempts() {
|
||||
return maxAttempts;
|
||||
}
|
||||
|
||||
public void setMaxAttempts(int maxAttempts) {
|
||||
this.maxAttempts = maxAttempts;
|
||||
}
|
||||
|
||||
public boolean isDlqEnabled() {
|
||||
return dlqEnabled;
|
||||
}
|
||||
|
||||
public void setDlqEnabled(boolean dlqEnabled) {
|
||||
this.dlqEnabled = dlqEnabled;
|
||||
}
|
||||
|
||||
public int getStreamMaxLen() {
|
||||
return streamMaxLen;
|
||||
}
|
||||
|
||||
public void setStreamMaxLen(int streamMaxLen) {
|
||||
this.streamMaxLen = streamMaxLen;
|
||||
}
|
||||
|
||||
}
|
||||
+30
@@ -0,0 +1,30 @@
|
||||
package org.dromara.common.bus.redis;
|
||||
|
||||
import org.redisson.api.RedissonClient;
|
||||
import org.springframework.boot.health.contributor.Health;
|
||||
import org.springframework.boot.health.contributor.HealthIndicator;
|
||||
|
||||
/**
|
||||
* Basic Redis bus health indicator.
|
||||
*
|
||||
* @author dromara
|
||||
*/
|
||||
public class RedisBusHealthIndicator implements HealthIndicator {
|
||||
|
||||
private final RedissonClient redissonClient;
|
||||
|
||||
public RedisBusHealthIndicator(RedissonClient redissonClient) {
|
||||
this.redissonClient = redissonClient;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Health health() {
|
||||
try {
|
||||
redissonClient.getBucket("spring:cloud:bus:health").isExists();
|
||||
return Health.up().withDetail("binder", "redis-stream").build();
|
||||
} catch (Exception e) {
|
||||
return Health.down(e).withDetail("binder", "redis-stream").build();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
+380
@@ -0,0 +1,380 @@
|
||||
package org.dromara.common.bus.redis;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.dromara.common.bus.config.RedisBusProperties;
|
||||
import org.redisson.api.RStream;
|
||||
import org.redisson.api.RedissonClient;
|
||||
import org.redisson.api.stream.AutoClaimResult;
|
||||
import org.redisson.api.stream.StreamAddArgs;
|
||||
import org.redisson.api.stream.StreamCreateGroupArgs;
|
||||
import org.redisson.api.stream.StreamMessageId;
|
||||
import org.redisson.api.stream.StreamReadGroupArgs;
|
||||
import org.springframework.cloud.stream.binder.AbstractBinder;
|
||||
import org.springframework.cloud.stream.binder.Binding;
|
||||
import org.springframework.cloud.stream.binder.ConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.ProducerProperties;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.MessageHandler;
|
||||
import org.springframework.messaging.SubscribableChannel;
|
||||
import org.springframework.messaging.support.MessageBuilder;
|
||||
import org.springframework.core.env.Environment;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
import java.io.Serial;
|
||||
import java.io.Serializable;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ThreadFactory;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
/**
|
||||
* Minimal Redis Stream binder used by Spring Cloud Bus.
|
||||
*
|
||||
* @author dromara
|
||||
*/
|
||||
@Slf4j
|
||||
public class RedisMessageChannelBinder extends AbstractBinder<MessageChannel, ConsumerProperties, ProducerProperties> {
|
||||
|
||||
private static final String MESSAGE_FIELD = "message";
|
||||
|
||||
private final RedissonClient redissonClient;
|
||||
|
||||
private final RedisBusProperties properties;
|
||||
|
||||
public RedisMessageChannelBinder(RedissonClient redissonClient, RedisBusProperties properties) {
|
||||
this.redissonClient = redissonClient;
|
||||
this.properties = properties;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Binding<MessageChannel> doBindConsumer(String name, String group, MessageChannel inboundBindTarget, ConsumerProperties consumerProperties) {
|
||||
Assert.notNull(inboundBindTarget, "inboundBindTarget must not be null");
|
||||
RStream<String, Object> stream = redissonClient.getStream(buildStreamName(name));
|
||||
String groupName = buildGroupName(name);
|
||||
String consumerName = buildConsumerName(name);
|
||||
createGroupIfNecessary(stream, groupName);
|
||||
stream.createConsumer(groupName, consumerName);
|
||||
RStream<String, Object> dlqStream = redissonClient.getStream(buildDlqStreamName(name, groupName));
|
||||
RedisConsumerBinding binding = new RedisConsumerBinding(name, inboundBindTarget, stream, dlqStream, groupName, consumerName, properties);
|
||||
binding.start();
|
||||
log.info("Redis bus consumer bound to stream: {}, group: {}", name, groupName);
|
||||
return binding;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Binding<MessageChannel> doBindProducer(String name, MessageChannel outboundBindTarget, ProducerProperties producerProperties) {
|
||||
Assert.isInstanceOf(SubscribableChannel.class, outboundBindTarget, "Redis bus producer requires a SubscribableChannel");
|
||||
RStream<String, Object> stream = redissonClient.getStream(buildStreamName(name));
|
||||
MessageHandler handler = message -> stream.add(StreamAddArgs.<String, Object>entry(
|
||||
MESSAGE_FIELD, new RedisBusMessage(message.getPayload(), new HashMap<>(message.getHeaders())))
|
||||
.trimNonStrict().maxLen(properties.getStreamMaxLen()).noLimit());
|
||||
((SubscribableChannel) outboundBindTarget).subscribe(handler);
|
||||
log.info("Redis bus producer bound to stream: {}", name);
|
||||
return new RedisProducerBinding(name, outboundBindTarget, (SubscribableChannel) outboundBindTarget, handler);
|
||||
}
|
||||
|
||||
private String buildStreamName(String destination) {
|
||||
return "spring:cloud:bus:" + (StringUtils.hasText(destination) ? destination : "springCloudBus");
|
||||
}
|
||||
|
||||
private String buildDlqStreamName(String destination, String groupName) {
|
||||
return buildStreamName(destination) + ":dlq:" + groupName;
|
||||
}
|
||||
|
||||
private String buildGroupName(String destination) {
|
||||
Environment environment = getApplicationContext().getEnvironment();
|
||||
String busId = environment.getProperty("spring.cloud.bus.id");
|
||||
if (!StringUtils.hasText(busId)) {
|
||||
busId = environment.getProperty("spring.application.name", "application");
|
||||
}
|
||||
return (StringUtils.hasText(destination) ? destination : "springCloudBus") + ":" + busId;
|
||||
}
|
||||
|
||||
private String buildConsumerName(String destination) {
|
||||
return (StringUtils.hasText(destination) ? destination : "springCloudBus") + ":" + buildInstanceId(getApplicationContext().getEnvironment());
|
||||
}
|
||||
|
||||
private String buildInstanceId(Environment environment) {
|
||||
String instanceId = firstText(
|
||||
environment.getProperty("spring.cloud.bus.redis.instance-id"),
|
||||
environment.getProperty("spring.cloud.nacos.discovery.ip"),
|
||||
environment.getProperty("spring.cloud.client.ip-address"),
|
||||
environment.getProperty("HOSTNAME"),
|
||||
environment.getProperty("COMPUTERNAME"),
|
||||
"localhost"
|
||||
);
|
||||
String port = firstText(
|
||||
environment.getProperty("server.port"),
|
||||
environment.getProperty("local.server.port"),
|
||||
"0"
|
||||
);
|
||||
return instanceId + ":" + port;
|
||||
}
|
||||
|
||||
private String firstText(String... values) {
|
||||
for (String value : values) {
|
||||
if (StringUtils.hasText(value)) {
|
||||
return value;
|
||||
}
|
||||
}
|
||||
return "";
|
||||
}
|
||||
|
||||
private void createGroupIfNecessary(RStream<String, Object> stream, String groupName) {
|
||||
try {
|
||||
stream.createGroup(StreamCreateGroupArgs.name(groupName).id(StreamMessageId.NEWEST).makeStream());
|
||||
} catch (Exception e) {
|
||||
if (!String.valueOf(e.getMessage()).contains("BUSYGROUP")) {
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public static class RedisBusMessage implements Serializable {
|
||||
|
||||
@Serial
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
private Object payload;
|
||||
|
||||
private Map<String, Object> headers = Map.of();
|
||||
|
||||
private int attempts;
|
||||
|
||||
private String lastError;
|
||||
|
||||
public RedisBusMessage() {
|
||||
}
|
||||
|
||||
public RedisBusMessage(Object payload, Map<String, Object> headers) {
|
||||
this.payload = payload;
|
||||
this.headers = headers == null ? Map.of() : headers;
|
||||
}
|
||||
|
||||
public Object getPayload() {
|
||||
return payload;
|
||||
}
|
||||
|
||||
public void setPayload(Object payload) {
|
||||
this.payload = payload;
|
||||
}
|
||||
|
||||
public Map<String, Object> getHeaders() {
|
||||
return headers;
|
||||
}
|
||||
|
||||
public void setHeaders(Map<String, Object> headers) {
|
||||
this.headers = headers == null ? Map.of() : headers;
|
||||
}
|
||||
|
||||
public int getAttempts() {
|
||||
return attempts;
|
||||
}
|
||||
|
||||
public void setAttempts(int attempts) {
|
||||
this.attempts = attempts;
|
||||
}
|
||||
|
||||
public String getLastError() {
|
||||
return lastError;
|
||||
}
|
||||
|
||||
public void setLastError(String lastError) {
|
||||
this.lastError = lastError;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private abstract static class AbstractRedisBinding implements Binding<MessageChannel> {
|
||||
|
||||
private final String name;
|
||||
private final MessageChannel target;
|
||||
private final AtomicBoolean running = new AtomicBoolean(false);
|
||||
|
||||
private AbstractRedisBinding(String name, MessageChannel target) {
|
||||
this.name = name;
|
||||
this.target = target;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void start() {
|
||||
running.set(true);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stop() {
|
||||
running.set(false);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isRunning() {
|
||||
return running.get();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void unbind() {
|
||||
stop();
|
||||
}
|
||||
|
||||
protected MessageChannel getTarget() {
|
||||
return target;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private static class RedisConsumerBinding extends AbstractRedisBinding {
|
||||
|
||||
private final RStream<String, Object> stream;
|
||||
private final RStream<String, Object> dlqStream;
|
||||
private final String groupName;
|
||||
private final String consumerName;
|
||||
private final RedisBusProperties properties;
|
||||
private final ExecutorService executor;
|
||||
|
||||
private RedisConsumerBinding(String name, MessageChannel target, RStream<String, Object> stream, RStream<String, Object> dlqStream,
|
||||
String groupName, String consumerName, RedisBusProperties properties) {
|
||||
super(name, target);
|
||||
this.stream = stream;
|
||||
this.dlqStream = dlqStream;
|
||||
this.groupName = groupName;
|
||||
this.consumerName = consumerName;
|
||||
this.properties = properties;
|
||||
ThreadFactory threadFactory = task -> {
|
||||
Thread thread = new Thread(task, "redis-bus-stream-" + name);
|
||||
thread.setDaemon(true);
|
||||
return thread;
|
||||
};
|
||||
this.executor = Executors.newSingleThreadExecutor(threadFactory);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void start() {
|
||||
if (isRunning()) {
|
||||
return;
|
||||
}
|
||||
super.start();
|
||||
executor.execute(this::poll);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stop() {
|
||||
super.stop();
|
||||
executor.shutdownNow();
|
||||
}
|
||||
|
||||
private void poll() {
|
||||
while (isRunning() && !Thread.currentThread().isInterrupted()) {
|
||||
try {
|
||||
claimPendingMessages();
|
||||
Map<StreamMessageId, Map<String, Object>> messages = stream.readGroup(
|
||||
groupName, consumerName, StreamReadGroupArgs.neverDelivered()
|
||||
.count(properties.getReadCount())
|
||||
.timeout(properties.getReadTimeout()));
|
||||
processMessages(messages);
|
||||
} catch (Exception e) {
|
||||
if (isRunning()) {
|
||||
log.error("Redis bus stream consume failed, group: {}", groupName, e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void claimPendingMessages() {
|
||||
AutoClaimResult<String, Object> result = stream.autoClaim(groupName, consumerName,
|
||||
properties.getClaimIdleTimeout().toMillis(), TimeUnit.MILLISECONDS,
|
||||
StreamMessageId.MIN, properties.getClaimCount());
|
||||
processMessages(result.getMessages());
|
||||
}
|
||||
|
||||
private void processMessages(Map<StreamMessageId, Map<String, Object>> messages) {
|
||||
if (messages == null || messages.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
for (Map.Entry<StreamMessageId, Map<String, Object>> entry : messages.entrySet()) {
|
||||
RedisBusMessage message = (RedisBusMessage) entry.getValue().get(MESSAGE_FIELD);
|
||||
if (message == null) {
|
||||
stream.ack(groupName, entry.getKey());
|
||||
continue;
|
||||
}
|
||||
processMessage(entry.getKey(), message);
|
||||
}
|
||||
}
|
||||
|
||||
private void processMessage(StreamMessageId messageId, RedisBusMessage message) {
|
||||
try {
|
||||
Message<Object> springMessage = MessageBuilder.withPayload(message.getPayload())
|
||||
.copyHeaders(message.getHeaders())
|
||||
.build();
|
||||
getTarget().send(springMessage);
|
||||
stream.ack(groupName, messageId);
|
||||
} catch (Exception e) {
|
||||
handleFailure(messageId, message, e);
|
||||
}
|
||||
}
|
||||
|
||||
private void handleFailure(StreamMessageId messageId, RedisBusMessage message, Exception e) {
|
||||
int attempts = message.getAttempts() + 1;
|
||||
message.setAttempts(attempts);
|
||||
message.setLastError(e.getClass().getName() + ": " + e.getMessage());
|
||||
if (attempts >= properties.getMaxAttempts()) {
|
||||
if (properties.isDlqEnabled()) {
|
||||
dlqStream.add(StreamAddArgs.<String, Object>entry(MESSAGE_FIELD, message)
|
||||
.trimNonStrict().maxLen(properties.getStreamMaxLen()).noLimit());
|
||||
}
|
||||
stream.ack(groupName, messageId);
|
||||
log.error("Redis bus message exhausted retries, group: {}, attempts: {}", groupName, attempts, e);
|
||||
return;
|
||||
}
|
||||
stream.add(StreamAddArgs.<String, Object>entry(MESSAGE_FIELD, message)
|
||||
.trimNonStrict().maxLen(properties.getStreamMaxLen()).noLimit());
|
||||
stream.ack(groupName, messageId);
|
||||
log.warn("Redis bus message retry scheduled, group: {}, attempts: {}", groupName, attempts, e);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void unbind() {
|
||||
try {
|
||||
stream.removeConsumer(groupName, consumerName);
|
||||
} catch (Exception e) {
|
||||
log.debug("Remove Redis stream consumer failed, group: {}, consumer: {}", groupName, consumerName, e);
|
||||
}
|
||||
stop();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isInput() {
|
||||
return true;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private static class RedisProducerBinding extends AbstractRedisBinding {
|
||||
|
||||
private final SubscribableChannel channel;
|
||||
private final MessageHandler handler;
|
||||
|
||||
private RedisProducerBinding(String name, MessageChannel target, SubscribableChannel channel, MessageHandler handler) {
|
||||
super(name, target);
|
||||
this.channel = channel;
|
||||
this.handler = handler;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void unbind() {
|
||||
channel.unsubscribe(handler);
|
||||
super.unbind();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
+1
@@ -1 +1,2 @@
|
||||
org.dromara.common.bus.config.BusCustomConfiguration
|
||||
org.dromara.common.bus.config.RedisBusBinderConfiguration
|
||||
|
||||
Reference in New Issue
Block a user