kafka优化之多线程消费

末蓝、 2022-03-17 09:07 170阅读 0赞

我之前用ssm消费kafka，在注解配置了消费队列，这是一种来生产多少，消费多少的形式，为了提升性能，加了一些优化：消费分为两部分：接收生产消费不变，执行消费行为用线程跑，但又不能无限制的开启线程，所以，最好用线程池管理

先看不需要返回结果的（runnable），应用：把消费代码封装，实现runnable接口，用线程池跑即可

public class ThreadPoolManager {
    
    	private static ThreadPoolManager sThreadPoolManager = new ThreadPoolManager();
    
    	// 线程池维护线程的最少数量
    	private static final int SIZE_CORE_POOL = 30;
    
    	// 线程池维护线程的最大数量
    	private static final int SIZE_MAX_POOL = 100;
    
    	// 线程池维护线程所允许的空闲时间
    	private static final int TIME_KEEP_ALIVE = 5000;
    
    	/* 
    	 * 单例
    	 */
    	public static ThreadPoolManager singleInstance() {
    		return sThreadPoolManager;
    	}
    
    	private final ThreadPoolExecutor mThreadPool = new ThreadPoolExecutor(SIZE_CORE_POOL, SIZE_MAX_POOL,
    			TIME_KEEP_ALIVE, TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>());
    
    	private ThreadPoolManager() {
    	}
    
    	public void perpare() {
    		if (mThreadPool.isShutdown() && !mThreadPool.prestartCoreThread()) {
    			@SuppressWarnings("unused")
    			int startThread = mThreadPool.prestartAllCoreThreads();
    		}
    	}
    
    	public void addExecuteTask(Runnable task) {
    		if (task != null) {
    			mThreadPool.execute(task);
    		}
    	}
    
    	protected boolean isTaskEnd() {
    		if (mThreadPool.getActiveCount() == 0) {
    			return true;
    		} else {
    			return false;
    		}
    	}
    
    	public void shutdown() {
    		mThreadPool.shutdown();
    	}
    }

如果是需要返回结果的，建议用callable，应用：把消费代码封装，实现callable接口，用线程池跑即可

public class CallableThreadPoolManager {
    
    	private static ExecutorService newFixedThreadPool;
    	
    	// 私有构造
        private ThreadPoolCallManager() {
        	newFixedThreadPool = Executors.newFixedThreadPool(Integer.parseInt(PropertiesUtil.getProperties("download_thread_limit")));
        }
    
        private static ThreadPoolCallManager single = new ThreadPoolCallManager();
    
        // 静态工厂方法
        public static ThreadPoolCallManager getInstance() {
            return single;
        }
    	
    	public Future<String> submit(CallableFeature task) {
    		return newFixedThreadPool.submit(task);
    	}
    	
    	public void shutdown() {
    		newFixedThreadPool.shutdown();
    	}
    }