package com.goafanti.easemob.queue; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.BoundListOperations; import org.springframework.data.redis.core.RedisConnectionUtils; import org.springframework.data.redis.core.RedisTemplate; import com.goafanti.easemob.bo.EasemobInfo; public class EasemobRedisQueue implements InitializingBean, DisposableBean { private RedisTemplate redisTemplate; private String key; private RedisConnectionFactory factory; private RedisConnection connection; private BoundListOperations listOperations; private Lock lock = new ReentrantLock(); private RedisQueueListener listener; private Thread listenerThread; private boolean isClosed; public void setRedisTemplate(RedisTemplate redisTemplate) { this.redisTemplate = redisTemplate; } public void setListener(RedisQueueListener listener) { this.listener = listener; } public void setKey(String key) { this.key = key; } @Override public void afterPropertiesSet() throws Exception { // do nothing /*factory = redisTemplate.getConnectionFactory(); connection = RedisConnectionUtils.getConnection(factory); listOperations = redisTemplate.boundListOps(key); if (listener != null) { listenerThread = new ListenerThread(); listenerThread.setDaemon(true); listenerThread.start(); }*/ } /** * blocking remove and get last item from queue:BRPOP * * @return */ public Object takeFromTail(int timeout) throws InterruptedException { lock.lockInterruptibly(); try { return listOperations.rightPop(timeout, TimeUnit.SECONDS); } finally { lock.unlock(); } } public Object takeFromTail() throws InterruptedException { return takeFromTail(0); } /** * 从队列的头,插入 */ public void pushFromHead(Object value) { listOperations.leftPush(value); } public void pushFromTail(Object value) { listOperations.rightPush(value); } /** * noblocking * * @return null if no item in queue */ public Object removeFromHead() { return listOperations.leftPop(); } public Object removeFromTail() { return listOperations.rightPop(); } /** * blocking remove and get first item from queue:BLPOP * * @return */ public Object takeFromHead(int timeout) throws InterruptedException { lock.lockInterruptibly(); try { return listOperations.leftPop(timeout, TimeUnit.SECONDS); } finally { lock.unlock(); } } public Object takeFromHead() throws InterruptedException { return takeFromHead(0); } @Override public void destroy() throws Exception { if (isClosed) { return; } shutdown(); RedisConnectionUtils.releaseConnection(connection, factory); } private void shutdown() { try { listenerThread.interrupt(); } catch (Exception e) { } } private class ListenerThread extends Thread { @Override public void run() { try { while (true) { Object value = takeFromHead(); // 逐个执行 if (value != null) { try { listener.onMessage((EasemobInfo) value); } catch (Exception e) { } } } } catch (InterruptedException e) { } } } }