重试机制（Guava Retry）和（spring-retry）、观察者模式、内部消费者生产者工具

ゝ一纸荒年。 2022-10-27 02:28 168阅读 0赞

**声明**：`详情请参考注释`

## 一、重试机制（Guava Retry）和（spring-retry） ##

import com.github.rholder.retry.*;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.remoting.RemoteAccessException;
    import org.springframework.retry.annotation.Backoff;
    import org.springframework.retry.annotation.Recover;
    import org.springframework.retry.annotation.Retryable;
    import org.springframework.util.StringUtils;
    
    import java.sql.SQLException;
    import java.time.LocalTime;
    import java.util.concurrent.TimeUnit;
    
    @Slf4j
    public class RetryExample { 
    
        /** * Retryable 的方法里面不能使用try...catch包裹，要在方法上抛出异常，不然不会触发。 * Spring为我们提供了原生的重试类库 * 在启动类或者配置类上添加@EnableRetry注解，并在需要重试的方法上添加@Retryable注解 * value：指定发生的异常进行重试 * include：和value一样，默认空，当exclude也为空时，所有异常都重试 * exclude：指定异常不重试，默认空，当include也为空时，所有异常都重试 * maxAttempts：重试次数，默认3 * backoff：重试补偿机制，默认没有 * delay:指定延迟后重试 * multiplier:指定延迟的倍数，默认为1。比如delay=5000l,multiplier=2时，第一次重试为5秒后，第二次为5*2=10秒，第三次为5*2*2=20秒 */
        @Retryable(value = { RemoteAccessException.class}, maxAttempts = 4)
        public String springRetry() { 
            long times = System.currentTimeMillis();
            log.info("hello times:{}", times);
            if (times % 4 != 0){ 
                log.error("发生异常，time：{}", LocalTime.now() );
                throw new RemoteAccessException("发生Hello异常");
            }
            return "hello " + times;
        }
    
        @Recover
        public void recover(RemoteAccessException e) { 
            log.info(e.getMessage());
        }
    
    
        /** * delay：如果不设置的话默认是1秒 * maxDelay：最大重试等待时间 * multiplier：用于计算下一个延迟时间的乘数(大于0生效) * random：随机重试等待时间(一般不用) * 第一，由于Spring Retry用到了Aspect增强，所以就会有使用Aspect不可避免的坑——方法内部调用， * 如果被 @Retryable 注解的方法的调用方和被调用方处于同一个类中， * 那么重试将会失效； * 第二，Spring的重试机制只支持对异常进行捕获，而无法对返回值进行校验判断重试。 * 如果想要更灵活的重试策略可以考虑使用Guava Retry，也是一个不错的选择 */
        @Retryable(maxAttempts = 5,backoff = @Backoff(delay = 3000))
        public void retrySomething() throws Exception{ 
            log.info("print Something{} is called");
            throw new SQLException();
        }
    
    
    
        /** * Guava Retry具有更强的灵活性 * 先创建一个Retryer实例，然后使用这个实例对需要重试的方法进行调用， * 可以通过很多方法来设置重试机制， * 比如使用retryIfException来对所有异常进行重试， * 使用retryIfExceptionOfType方法来设置对指定异常进行重试， * 使用retryIfResult来对不符合预期的返回结果进行重试， * 使用retryIfRuntimeException方法来对所有RuntimeException进行重试 */
        public void guavaRetry() { 
            Retryer<String> retrys = RetryerBuilder.<String>newBuilder()
                    .retryIfException()
                    .retryIfResult(StringUtils::isEmpty)
                    .withWaitStrategy(WaitStrategies.fixedWait(3, TimeUnit.SECONDS))
                    .withStopStrategy(StopStrategies.stopAfterAttempt(3))
                    .build();
            try { 
                retrys.call(() -> { 
                    long times = System.currentTimeMillis();
                    log.info("hello times:{}", times);
                    if (times % 4 != 0){ 
                        log.error("发生异常，time：{}", LocalTime.now() );
                        throw new RemoteAccessException("发生Hello异常");
                    }
                    return "hello " + times;
                });
            } catch (Exception e){ 
                log.error("重试机制出现异常：{}",e);
            }
        }
    }

import lombok.extern.slf4j.Slf4j;
    import org.springframework.boot.CommandLineRunner;
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.retry.annotation.EnableRetry;
    import org.springframework.scheduling.annotation.EnableAsync;
    
    @Slf4j
    @EnableAsync
    @EnableRetry
    @SpringBootApplication
    public class StreammapApplication implements CommandLineRunner { 
    
        public static void main(String[] args) { 
            SpringApplication.run(StreammapApplication.class, args);
        }
    
        @Override
        public void run(String... args) { 
            log.info("启动服务...");
        }
    }

<dependency>
                <groupId>org.springframework.retry</groupId>
                <artifactId>spring-retry</artifactId>
            </dependency>
    
            <dependency>
                <groupId>org.aspectj</groupId>
                <artifactId>aspectjweaver</artifactId>
            </dependency>
    
            <dependency>
                <groupId>com.github.rholder</groupId>
                <artifactId>guava-retrying</artifactId>
                <version>2.0.0</version>
            </dependency>

## 二、观察者模式 ##

import lombok.extern.slf4j.Slf4j;
    import org.springframework.context.ApplicationEvent;
    
    @Slf4j
    public class UserRegisterEvent extends ApplicationEvent { 
        /** * 用户名 */
        private String username;
    
        /** * 传递参数或者对象 * @param source * @param username */
        public UserRegisterEvent(Object source, String username) { 
            super(source);
            this.username = username;
            log.info("执行此方法");
        }
    
        public String getUsername() { 
            return username;
        }
    }

import lombok.extern.slf4j.Slf4j;
    import org.jetbrains.annotations.NotNull;
    import org.springframework.context.ApplicationEventPublisher;
    import org.springframework.context.ApplicationEventPublisherAware;
    import org.springframework.stereotype.Service;
    
    @Slf4j
    @Service
    public class UserService implements ApplicationEventPublisherAware { 
    
        private ApplicationEventPublisher applicationEventPublisher;
    
        @Override
        public void setApplicationEventPublisher(@NotNull ApplicationEventPublisher eventPublisher) { 
            this.applicationEventPublisher = eventPublisher;
        }
    
        public void register(String username) { 
            // ... 执行注册逻辑
            log.info("[register][执行用户({}) 的注册逻辑]", username);
            // ... 发布
            applicationEventPublisher.publishEvent(new UserRegisterEvent(this, username));
        }
    }

import lombok.extern.slf4j.Slf4j;
    import org.springframework.context.event.EventListener;
    import org.springframework.stereotype.Service;
    
    @Slf4j
    @Service
    public class CouponService { 
    
        @EventListener
        public void addCoupon(UserRegisterEvent event) { 
            log.info("[addCoupon][给用户({}) 发放优惠劵]", event.getUsername());
        }
    }

import lombok.extern.slf4j.Slf4j;
    import org.springframework.context.ApplicationListener;
    import org.springframework.scheduling.annotation.Async;
    import org.springframework.stereotype.Service;
    
    @Slf4j
    @Service
    public class EmailService implements ApplicationListener<UserRegisterEvent> { 
    
        @Override
        @Async
        public void onApplicationEvent(UserRegisterEvent event) { 
            log.info("[onApplicationEvent][给用户({}) 发送邮件]", event.getUsername());
        }
    }

import lombok.extern.slf4j.Slf4j;
    import org.springframework.context.event.EventListener;
    import org.springframework.stereotype.Service;
    
    @Slf4j
    @Service
    public class RedEnvelopesService { 
    
        @EventListener
        public void addRed(UserRegisterEvent event) { 
            log.info("[addRed][给用户({}) 发放红包]", event.getUsername());
        }
    }

import lombok.extern.slf4j.Slf4j;
    import org.springframework.context.ApplicationListener;
    import org.springframework.scheduling.annotation.Async;
    import org.springframework.stereotype.Service;
    
    @Slf4j
    @Service
    public class ShortMessageService implements ApplicationListener<UserRegisterEvent>{ 
    
        @Override
        @Async
        public void onApplicationEvent(UserRegisterEvent event) { 
            log.info("[onApplicationEvent][给用户({}) 发送短信]", event.getUsername());
        }
    }

## 三、内部消费者生产者工具 ##

import com.lmax.disruptor.EventFactory;
    import com.lmax.disruptor.RingBuffer;
    import com.lmax.disruptor.YieldingWaitStrategy;
    import com.lmax.disruptor.dsl.Disruptor;
    import com.lmax.disruptor.dsl.ProducerType;
    
    public final class DisruptorUtil { 
    
        private static DisruptorUtil disruptorUtil;
    
        private Disruptor<MessageEvent> disruptor;
    
        /** * RingBuffer 大小，必须是 2 的 N 次方； * RingBuffer 的大小，数字为字节数 * RingBuffer 是框架启动器内部的缓存区，用来存储 event 内的 task 数据 */
        private static final int RING_BUFFER_SIZE = 1024 * 1024;
    
        private RingBuffer<MessageEvent> ringBuffer;
    
        private MessageEventProducer messageEventProducer;
    
        private final MessageEventConsumer messageEventConsumer;
    
        private DisruptorUtil() { 
            //在该框架中，所有的 task 的包装类被称为 Event，EventFactory 则是 Event 的生产者
            EventThreadFactory eventThreadFactory = new EventThreadFactory();
            EventFactory<MessageEvent> eventFactory = new MessageEventFactory();
            //创建一个 Disruptor 启动器，其中 DaemonThreadFactory 是一个线程工厂的实现类
            disruptor = new Disruptor<>(eventFactory, RING_BUFFER_SIZE, eventThreadFactory, ProducerType.SINGLE,
                    new YieldingWaitStrategy());
            //创建一个 Disruptor 启动器，其中 DaemonThreadFactory 是一个线程工厂的实现类
            //Disruptor<LongEvent> disruptor = new Disruptor<>(eventThreadFactory, RING_BUFFER_SIZE, DaemonThreadFactory.INSTANCE);
            //该框架本质上是 生产-消费 设计模式的应用。所有的消费者被冠名为 handler
            //handleEventsWith(...) 方法会在启动器中注册 handler
            //此处的参数是不定数量的，可以有多个消费者，每个消费者都可以获取 Event
            //disruptor.handleEventsWith(new LongEventHandler("handler1"),new LongEventHandler("handler2"))
            messageEventConsumer = new MessageEventConsumer();
            disruptor.handleEventsWith(messageEventConsumer);
        }
    
        /** * 获取 LogDisruptorUtil 实例 初始化 * * @return LogDisruptorUtil */
        public static DisruptorUtil getInstance() { 
            if (disruptorUtil == null) { 
                synchronized (DisruptorUtil.class) { 
                    if (disruptorUtil == null) { 
                        disruptorUtil = new DisruptorUtil();
                        return disruptorUtil;
                    }
                }
            }
            return disruptorUtil;
        }
        /** * 启动disruptor * 启动器开始执行，并获取其内部的缓存区 */
        public void start() { 
            disruptor.start();
            ringBuffer = disruptor.getRingBuffer();
            messageEventProducer = new MessageEventProducer(ringBuffer);
            //应用关闭前关闭disrupt
            Runtime.getRuntime().addShutdownHook(new Thread(() -> disruptor.shutdown()));
        }
    
        /** * 生产者发布事件 */
        public void push(String message) { 
            messageEventProducer.push(message);
        }
    
        public static void main(String[] args) { 
            //获取实例
            DisruptorUtil disruptorUtil = DisruptorUtil.getInstance();
            //启动
            disruptorUtil.start();
            //发布消息
            disruptorUtil.push("disruptor test !");
        }
    }

import org.jetbrains.annotations.NotNull;
    
    import java.util.concurrent.ThreadFactory;
    
    /** * 线程工厂，用于生产消费者线程 */
    public class EventThreadFactory implements ThreadFactory { 
        @Override
        public Thread newThread(@NotNull Runnable runnable) { 
            return new Thread(runnable);
        }
    }

/** * Event 类，本质上是数据的封装，是生产者和消费者之间进行数据传递的介质 */
    public class MessageEvent { 
    
        private String message;
    
        public String getMessage() { 
            return this.message;
        }
    
        public void setMessage(String message) { 
            this.message = message;
        }
    }

import com.lmax.disruptor.EventHandler;
    
    /** * 消费者，在此类实现具体的业务逻辑 *消费者，必须实现 Disruptor 自带的 EventHandler 接口 * */
    public class MessageEventConsumer implements EventHandler<MessageEvent> { 
    
        /** * 此方法为最终的消费 Event 的方法 * @param event * @param sequence * @param endOfBatch */
        @Override
        public void onEvent(MessageEvent event, long sequence, boolean endOfBatch) { 
            System.out.println("消费者消费消息：" + event.getMessage());
        }
    }

import com.lmax.disruptor.EventFactory;
    
    
    /** * 事件工厂，disruptor生产消息对象 * Event 的生产工厂类，必须实现 Disruptor 自带的 EventFactory 接口 */
    public class MessageEventFactory implements EventFactory<MessageEvent> { 
        @Override
        public MessageEvent newInstance() { 
            return new MessageEvent();
        }
    }

import com.lmax.disruptor.RingBuffer;
    
    
    /** * 生产者，在此类中向disruptor发布消息 * 生产者，主要负责往 RingBuffer 中写入数据 * 生产者类在框架中并非必须，但是一般情况下都会做一定程度的封装 */
    public class MessageEventProducer { 
    
        private final RingBuffer<MessageEvent> ringBuffer;
    
        /** * 生产者的构造器负责获取并存储启动器中的 RingBuffer * @param ringBuffer */
        public MessageEventProducer(RingBuffer<MessageEvent> ringBuffer) { 
            this.ringBuffer = ringBuffer;
        }
        /** * 将接收到的消息输出到ringBuffer */
        public void push(String message) { 
            //请求下一个事件序号；
            //sequence 是 RingBuffer 中的一个数据块，类似于一个数据地址
            long sequence = ringBuffer.next();
            try { 
                //获取该序号对应的事件对象；
                //用数据地址去获取到一个 Event 事件类实例
                MessageEvent event = ringBuffer.get(sequence);
                //在实例中存入 ByteBuffer 中的数据
                event.setMessage(message);
                System.out.println("生产者发布消息:" + message);
            } finally { 
                //发布事件；
                //发布该数据块，此时消费者们都可以看到该数据块了，可以进行消费
                ringBuffer.publish(sequence);
            }
        }
    }

<dependency>
                <groupId>com.lmax</groupId>
                <artifactId>disruptor</artifactId>
                <version>3.4.2</version>
            </dependency>
    
            <dependency>
                <groupId>cn.hutool</groupId>
                <artifactId>hutool-all</artifactId>
                <version>5.3.10</version>
            </dependency>

参考：[https://mp.weixin.qq.com/s/94oe5c\_7ouE1GbyiPfNg5g][https_mp.weixin.qq.com_s_94oe5c_7ouE1GbyiPfNg5g]

参考：[https://blog.csdn.net/qq\_22017479/article/details/90214749][https_blog.csdn.net_qq_22017479_article_details_90214749]

[https_mp.weixin.qq.com_s_94oe5c_7ouE1GbyiPfNg5g]: https://mp.weixin.qq.com/s/94oe5c_7ouE1GbyiPfNg5g
[https_blog.csdn.net_qq_22017479_article_details_90214749]: https://blog.csdn.net/qq_22017479/article/details/90214749