| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152 |
- 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<String, Object> redisTemplate;
- private String key;
- private RedisConnectionFactory factory;
- private RedisConnection connection;
- private BoundListOperations<String, Object> listOperations;
- private Lock lock = new ReentrantLock();
- private RedisQueueListener<EasemobInfo> listener;
- private Thread listenerThread;
- private boolean isClosed;
- public void setRedisTemplate(RedisTemplate<String, Object> redisTemplate) {
- this.redisTemplate = redisTemplate;
- }
- public void setListener(RedisQueueListener<EasemobInfo> 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) {
- }
- }
- }
- }
|