java concurrent 包之 ScheduledThreadPoolExecutor

客官°小女子只卖身不卖艺 2023-06-20 06:53 10阅读 0赞

java 自带定时任务的另一种实现

## 使用示例: ##

//package com.ws.com.util;
    
    import java.util.concurrent.ScheduledThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    /**
     * 定时执行 util
     */
    public class TimerUtil {
    
    
        /**
         * 可以看作强化后的Timer ,且不用继承Timer类
         */
        // 定义的线程池容量
        private static ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1000);
    
    
        /**
         * 设置定时执行
         * @param period  执行间隔
         * @param unit   时间单位
         *               SECONDS
         */
        public static void setTimer(Runnable command,long period, TimeUnit unit){
    //        scheduleWithFixedDelay  在...之后执行
    //        scheduleAtFixedRate    再...之前执行
            // command 线程实现类
            // initialDelay 初始延迟,第一次执行在程序启动多少秒后执行
            // period 执行的时间间隔,主要设置参数
            // unit 时间的单位(初始延迟和时间间隔的时间单位,TimeUnit 的枚举值
            executor.scheduleAtFixedRate(command,6,period, unit);
        }
    
    
    }

## 简单说明: ##

可以看作示Java 对Timer 类的复杂封装,引入了线程池的实现,

主要通过

scheduleAtFixedRate

scheduleWithFixedDelay

这两个方法来定义定时任务,需要传入一个实现 Runnable 接口的线程实现类,每个线程在定义的时间到了,就会去执行实现类的事务,但是**有报错就停止执行,且没有报错打印,**后来测试了一下发现再线程实现类,run 方法里面将需要处理的代码用try\{\}catch()\{\} 处理就可以,这个地方因为run 方法是实现Runnable 接口的,所以不能抛出,这里就算报错也会继续执行下去

## 问题: ##

如果执行任务的过程中出现异常或者其他问题,定时任务会停止执行,但是大部分时候都不会抛出异常,也不知道原因,这个待后面有空验证和查找解决办法

## 使用场景: ##

我们知道现在在大量的定时任务的时候,

一般项目首选的就是quartz ,引入和设置cron 表达式的方式,这个是一个重实现, 重实现设置导入复杂,有时候项目中没有配置定时任务,启动还会莫名报错,但是设置方便,操控性强,可配置在页面,实现启动,实现关门,实现修改定时任务的时间间隔,

或者通过spring 自带的定时任务处理方式,轻实现, 但是轻实现虽然是spring 自带的,也可以通过cron 表达式配置,配置简单,但是操控性差, 出了问题不好监控和调整,一旦配置好了,除非修改项目重启,不然不能调整定时任务的周期,启动,关闭

但是Java 自带的这种实现方式,方便我们在测试的时候,需要定时任务,但是没必要去配置quartz 和spring 环境,方便测试的时候简单实现测试需求,或者 处理一些高并发的需求的时候,可以设置大容量的线程和短的时间间隔,去调用,也可以简单的实现需求,比如消息队列的消息处理,当消息比较多的时候,单个线程处理,比如获取消息并处理消息保存在数据库或其他操作,可以通过先将消息放入 异步队列中,然后可以设置 1000个线程, 每隔 50 微秒 ,就去队列中取待获取的消息,并处理

## 例子: ##

说明一个rabbitmq 的例子,由于涉及隐私,某些数据已经脱敏,仅供展示

### 连接工厂,创建连接 ###

package mq.util;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * rabbitmq 连接工厂
     */
    public class AmqpConnectFactory {
    
    
        public static String username ="";
        public static String password ="";
        public static String inPlayPackageId = "";
    
        public static String inPlayHost = "";
        
    
        public static void main(String[] args) throws IOException, TimeoutException {
            ClientUitl.startInPlayMq(); // 启动mq
            inPlay();//连接mq 
    
        }
    
    
        /**
         *
         * @throws IOException
         * @throws TimeoutException
         */
        public static void inPlay() throws IOException, TimeoutException {
    //        createConnection(inPlayHost,inPlayPackageId,"C:\\Users\\admin\\Desktop\\newRecord\\");
            ConnectionFactory connectioinFactory = getConnectioinFactory(inPlayHost);
            Connection connection = getConnection(connectioinFactory);
    
            Channel channel =  getChannel(connection,inPlayPackageId,"C:\\Users\\admin\\Desktop\\newRecord\\"); // 开启一个通道并设置 相关的消费者等信息
        }
    
    
    
    
        /**
         * 得到连接工厂
         * @param host
         * @return
         * @throws IOException
         * @throws TimeoutException
         */
        private static ConnectionFactory getConnectioinFactory(String host) throws IOException, TimeoutException {
            ConnectionFactory connectionFactory = new ConnectionFactory();
    
            connectionFactory.setHost(host);
            connectionFactory.setPort(5672);
            connectionFactory.setUsername(username);
            connectionFactory.setPassword(password);
            connectionFactory.setAutomaticRecoveryEnabled(true);
            connectionFactory.setVirtualHost("Customers");
            connectionFactory.setRequestedHeartbeat(580);
            connectionFactory.setNetworkRecoveryInterval(1000);// 设置连接间隔
            // 或者使用 url 进行连接
            // factory.setUri（“ amqp：// userName：password @ hostName：portNumber / virtualHost”）;
    //        connectionFactory.setUri("amqp://"+username+":"+password+"@"+ inPlayHost +":" + port +"/" + virtualHost);
    
    //        final Connection conn = connectionFactory.newConnection(); // 新建一个连接
            return connectionFactory;
        }
    
    
        /**
         * 得到一个新的连接
         * @param factory
         * @return
         * @throws IOException
         * @throws TimeoutException
         */
        public static Connection getConnection(ConnectionFactory factory) throws IOException, TimeoutException {
            Connection connection = factory.newConnection();
            return connection;
        }
    
    
        /**
         * 得到一个新的通道
         * @param conn
         * @param packageId
         * @return
         * @throws IOException
         */
        public static Channel getChannel(Connection conn,String packageId,String path) throws IOException {
            final Channel channel = conn.createChannel(); // 打开一个通道(同文档中的模型)
    //        model.BasicQos(prefetchSize: 0, prefetchCount: 1000, global: false);
            channel.basicQos(0, 1000, false);// 配置服务质量
            // 订阅主题, 设置消息自动确认,配置消费者
            channel.basicConsume("_" + packageId +"_", true, new TestConsumer(channel,conn,packageId,path));
            return channel;
        }
    
    
    
    
    
    
    
    
    
    
    
    
    
        @Override
        protected void finalize() throws Throwable {
            super.finalize();
        }
    }

### 消费者配置 ###

说明: 该例实现了 Consumer 接口的handleConsumeOk() 方法,在消费者配置成功了,(调用定时任务)就调用上文代码中的线程池去队列中取消息, 具体消息处理方法handleDelivery () 方法不再处理具体事务,仅仅将消息放入异步队列中

package mq.util;
    
    import com.rabbitmq.client.*;
    import com.ws.com.util.TimerUtil;
    
    import java.io.IOException;
    import java.util.concurrent.TimeUnit;
    
    /**
     *   mq 消费者
     *
     */
    public class TestConsumer implements Consumer { //extends DefaultConsumer {
    
    
        private Channel channel;
        private Connection connection;
        private volatile String consumerTag;
    
    
        private String packageId;
        private String path;
    
    
        /**
         * Constructs a new instance and records its association to the passed-in channel.
         *
         * @param channel the channel to which this consumer is attached
         */
        public TestConsumer(Channel channel,Connection connection,String packageId,String path) {
    //        super(channel);
            this.channel = channel;
            this.connection = connection;
            this.packageId = packageId;
            this.path = path;
        }
    
        /**
         * 接收到此使用者的basic.deliver时调用
         * @param consumerTag
         * @param envelope
         * @param properties
         * @param body
         * @throws IOException
         */
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    //        super.handleDelivery(consumerTag, envelope, properties, body);
            String routingKey = envelope.getRoutingKey();
            String contentType = properties.getContentType();
            long deliveryTag = envelope.getDeliveryTag();
            // 处理消息,将消息放入队列中
            CacheUtil.queue.offer(new String(body));
    //        handleMessage(body,path);
    
            // 由于前面设置通道时已经设置了消息消费后自动确认,所以这里不用再确认消息,否则 双重确认会导致通道异常关闭
    //            channel.basicAck(deliveryTag, true);
    //
        }
    
    //
        /**
         * ：当通道或基础连接被关闭时调用。
         * 通道关闭的问题
         * @param consumerTag
         * @param sig
         */
        @Override
        public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
            System.out.println("通道意外关闭,重新设置通道");
            System.out.println("关闭原因"+sig.getMessage());
    //        this.chan
            try {
                Channel newChannel = connection.createChannel();
                channel = newChannel;// 将绑定的通道设置为新通道
                channel.basicQos(0, 1000, false);// 配置服务质量
                // 配置消费者
                channel.basicConsume("_" + packageId +"_", true, this);
            } catch (IOException e) {
                e.printStackTrace();
            }
    
    
        }
    
        /**
         * 当消费者由于除调用Channel.basicCancel之外的其他原因被取消时调用。
         * @param consumerTag
         * @throws IOException
         */
        @Override
        public void handleCancel(String consumerTag) throws IOException {
    //        super.handleCancel(consumerTag);
            this.consumerTag = consumerTag;
        }
    
        /**
         * 当消费者被Channel.basicCancel的调用取消时调用。
         * @param consumerTag
         */
        @Override
        public void handleCancelOk(String consumerTag) {
    //        super.handleCancelOk(consumerTag);
        }
    
        /**
         * 当使用者通过对任何通道的调用注册时调用 Channel.basicConsume方法。
         * @param consumerTag
         */
        @Override
        public void handleConsumeOk(String consumerTag) {
    //        super.handleConsumeOk(consumerTag);
            System.out.println("注册通过成功");
            // 开启线程池读取消息,并保存, 每 500毫秒调用1次,拿取一个消息
            TimerUtil.setTimer(new SaveRunner(path),50,TimeUnit.MICROSECONDS);
    
        }
    
    
        /**
         *  当 basic.recover. 收到 回复 basic.recover-ok  时调用
         * @param consumerTag
         */
        @Override
        public void handleRecoverOk(String consumerTag) {
            System.out.println("接收消息成功");
    //        super.handleRecoverOk(consumerTag);
        }
    
    
        // ==============================================================
    
    
        public Channel getChannel() {
            return channel;
        }
    
        public void setChannel(Channel channel) {
            this.channel = channel;
        }
    
        public Connection getConnection() {
            return connection;
        }
    
        public void setConnection(Connection connection) {
            this.connection = connection;
        }
    
        public String getConsumerTag() {
            return consumerTag;
        }
    
    
        // ================================================= 具体处理消息的逻辑 ==================================================================================================================
    
    
    
    
    }

### 异步队列 ###

package mq.util;
    
    import com.alibaba.dubbo.common.utils.ConcurrentHashSet;
    import mq.event.MatchInfo;
    
    import java.util.Map;
    import java.util.Set;
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.LinkedBlockingQueue;
    
    /**
     *
     *  缓存信息map
     *
     */
    public class CacheUtil {
    
    
        /**
         * 这里 存消息 offer,取消息 poll ,即使队列中无值也不会报错
         */
        public static LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>();
    
    
       
    }

### 具体处理消息的线程实现类 ###

package mq.util;
    
    import org.apache.commons.lang3.StringUtils;
    
    /**
     * 具体消息处理类
     */
    public class SaveRunner implements Runnable {
    
        //    private volatile String message;
        private String path;
    
        public SaveRunner(String path) {
            this.path = path;
        }
    
        @Override
        public void run() {
            // 从队列取消息
            try{
                String message = CacheUtil.queue.poll();
                if (StringUtils.isNotEmpty(message)) {
                handleMessage(message, path);
                }
            }catch(Exception e) {
                e.printStackTrace();
            }
    
    
        }
    
    
        /**
         * 保存到文件中,将消息放入队列中
         *
         * @param s
         * @param path
         */
        private void handleMessage(String s, String path) {
    //    处理消息   
        }
    
    }