EasemobRedisQueue.java 3.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152
  1. package com.goafanti.easemob.queue;
  2. import java.util.concurrent.TimeUnit;
  3. import java.util.concurrent.locks.Lock;
  4. import java.util.concurrent.locks.ReentrantLock;
  5. import org.springframework.beans.factory.DisposableBean;
  6. import org.springframework.beans.factory.InitializingBean;
  7. import org.springframework.data.redis.connection.RedisConnection;
  8. import org.springframework.data.redis.connection.RedisConnectionFactory;
  9. import org.springframework.data.redis.core.BoundListOperations;
  10. import org.springframework.data.redis.core.RedisConnectionUtils;
  11. import org.springframework.data.redis.core.RedisTemplate;
  12. import com.goafanti.easemob.bo.EasemobInfo;
  13. public class EasemobRedisQueue implements InitializingBean, DisposableBean {
  14. private RedisTemplate<String, Object> redisTemplate;
  15. private String key;
  16. private RedisConnectionFactory factory;
  17. private RedisConnection connection;
  18. private BoundListOperations<String, Object> listOperations;
  19. private Lock lock = new ReentrantLock();
  20. private RedisQueueListener<EasemobInfo> listener;
  21. private Thread listenerThread;
  22. private boolean isClosed;
  23. public void setRedisTemplate(RedisTemplate<String, Object> redisTemplate) {
  24. this.redisTemplate = redisTemplate;
  25. }
  26. public void setListener(RedisQueueListener<EasemobInfo> listener) {
  27. this.listener = listener;
  28. }
  29. public void setKey(String key) {
  30. this.key = key;
  31. }
  32. @Override
  33. public void afterPropertiesSet() throws Exception {
  34. // do nothing
  35. /*factory = redisTemplate.getConnectionFactory();
  36. connection = RedisConnectionUtils.getConnection(factory);
  37. listOperations = redisTemplate.boundListOps(key);
  38. if (listener != null) {
  39. listenerThread = new ListenerThread();
  40. listenerThread.setDaemon(true);
  41. listenerThread.start();
  42. }*/
  43. }
  44. /**
  45. * blocking remove and get last item from queue:BRPOP
  46. *
  47. * @return
  48. */
  49. public Object takeFromTail(int timeout) throws InterruptedException {
  50. lock.lockInterruptibly();
  51. try {
  52. return listOperations.rightPop(timeout, TimeUnit.SECONDS);
  53. } finally {
  54. lock.unlock();
  55. }
  56. }
  57. public Object takeFromTail() throws InterruptedException {
  58. return takeFromTail(0);
  59. }
  60. /**
  61. * 从队列的头,插入
  62. */
  63. public void pushFromHead(Object value) {
  64. listOperations.leftPush(value);
  65. }
  66. public void pushFromTail(Object value) {
  67. listOperations.rightPush(value);
  68. }
  69. /**
  70. * noblocking
  71. *
  72. * @return null if no item in queue
  73. */
  74. public Object removeFromHead() {
  75. return listOperations.leftPop();
  76. }
  77. public Object removeFromTail() {
  78. return listOperations.rightPop();
  79. }
  80. /**
  81. * blocking remove and get first item from queue:BLPOP
  82. *
  83. * @return
  84. */
  85. public Object takeFromHead(int timeout) throws InterruptedException {
  86. lock.lockInterruptibly();
  87. try {
  88. return listOperations.leftPop(timeout, TimeUnit.SECONDS);
  89. } finally {
  90. lock.unlock();
  91. }
  92. }
  93. public Object takeFromHead() throws InterruptedException {
  94. return takeFromHead(0);
  95. }
  96. @Override
  97. public void destroy() throws Exception {
  98. if (isClosed) {
  99. return;
  100. }
  101. shutdown();
  102. RedisConnectionUtils.releaseConnection(connection, factory);
  103. }
  104. private void shutdown() {
  105. try {
  106. listenerThread.interrupt();
  107. } catch (Exception e) {
  108. }
  109. }
  110. private class ListenerThread extends Thread {
  111. @Override
  112. public void run() {
  113. try {
  114. while (true) {
  115. Object value = takeFromHead();
  116. // 逐个执行
  117. if (value != null) {
  118. try {
  119. listener.onMessage((EasemobInfo) value);
  120. } catch (Exception e) {
  121. }
  122. }
  123. }
  124. } catch (InterruptedException e) {
  125. }
  126. }
  127. }
  128. }