MessageRedisQueue.java 3.8 KB

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