1 public class AlarmQueue<T> implements InitializingBean, DisposableBean, Component {
  2     private static final CLogger logger = Utils.getLogger(AlarmQueue.class);
  3
  4     @Autowired
  5     ThreadFacade thf;
  6     @Autowired
  7     private CloudBus bus;
  8
  9     private RedisTemplate redisTemplate;
 10     private String key;
 11     private RedisConnectionFactory factory;
 12     private RedisConnection connection;
 13     private Lock lock = new ReentrantLock();//基于底层IO阻塞考虑
 14     private Thread alarmThread;
 15     private boolean isInit = true;
 16     private boolean isClosed;
 17
 18     public void setRedisTemplate(RedisTemplate redisTemplate) {
 19         this.redisTemplate = redisTemplate;
 20     }
 21
 22     public void setKey(String key) {
 23         this.key = key;
 24     }
 25
 26     @Override
 27     public void afterPropertiesSet() throws Exception {//
 28         logger.info("============== afterPropertiesSet =================");
 29         factory = redisTemplate.getConnectionFactory();
 30         connection = RedisConnectionUtils.getConnection(factory);
 31         alarmThread = new AlarmThread();
 32         alarmThread.setDaemon(true);
 33         alarmThread.start();
 34     }
 35
 36     class AlarmThread extends Thread {
 37         @Override
 38         public void run() {
 39             try {
 40                 if (isInit) {
 41                     Thread.sleep(30000);
 42                     isInit = false;
 43                 }
 44
 45                 logger.info("============== AlarmThread Start =================");
 46                 while (true) {
 47                     T value = takeFromTail(0);
 48                     if (value != null) {
 49                         try {
 50                             // listener.onMessage(value);
 51                             HandleAlarmMsg amsg = new HandleAlarmMsg();
 52                             amsg.setAlarmValue(value.toString());
 53                             bus.makeTargetServiceIdByResourceUuid(amsg, AlarmConstant.SERVICE_ID_ALARM_LOG, Platform.getUuid());
 54                             bus.send(amsg);
 55                         } catch (Exception e) {
 56                             logger.error(String.format("fail to handle alarm!Alarm content: %s, Error: %s", value.toString(), e.getMessage()));
 57                         }
 58                     }
 59                 }
 60             } catch (InterruptedException e) {
 61                 throw new RuntimeException(String.format("Alarm thread InterruptedException! Error: %s", e.getMessage()));
 62             }
 63         }
 64     }
 65
 66     public T takeFromTail(int timeout) throws InterruptedException {
 67         lock.lockInterruptibly();
 68         try {
 69             List<byte[]> results = connection.bRPop(timeout, key.getBytes());
 70             if (CollectionUtils.isEmpty(results)) {
 71                 return null;
 72             }
 73             return (T) redisTemplate.getValueSerializer().deserialize(results.get(1));
 74         } catch (Exception e) {
 75             throw new RuntimeException(String.format("fail to take value from queue %s! Error: ", this.key, e.getMessage()));
 76         } finally {
 77             lock.unlock();
 78         }
 79     }
 80
 81     @Override
 82     public void destroy() throws Exception {
 83         if (isClosed) {
 84             return;
 85         }
 86         shutdown();
 87         RedisConnectionUtils.releaseConnection(connection, factory);
 88         isClosed = true;
 89     }
 90
 91     private void shutdown() {
 92         AlarmThread.interrupted();
 93     }
 94
 95     @Override
 96     public boolean start() {
 97         return true;
 98     }
 99
100     @Override
101     public boolean stop() {
102         return true;
103     }
104 }

 1 public class AlarmLogManagerImpl extends AbstractService implements ApiMessageInterceptor {
 2
 3     private static final CLogger logger = Utils.getLogger(AlarmLogManagerImpl.class);
 4     @Autowired
 5     private CloudBus bus;
 6     @Autowired
 7     private DatabaseFacade dbf;
 8     @Autowired
 9     private ThreadFacade thf;
10     @Autowired
11     private RESTFacade restf;
12     @Autowired
13     private SmsService smsService;
14     @Autowired
15     private MailService mailService;
16
17   ....
18 }

<bean id="AlarmLogManager" class="com.syscxp.alarm.log.AlarmLogManagerImpl"init-method="AlarmLogManager" destroy-method="destroy"> .... 

</bean>