redis非同期メッセージ処理の実装

53157 ワード
ジェーディs blpop rpush 非同期メッセージ
詳細
 
メッセージのサブスクリプションとパブリケーションをredisで実装する方法
 
テストクラス
/** *redis */ @Controller @RequestMapping(value="/queue") publicclassTestControlller{
  • @Autowired
    • privateRedisTemplate redisService;
  • static{
    • newTestControlller().initThread("msgQueue"); }
  • /** - **/
    • public void initThread(String key){ newThread(new blpopMessageHandler(key)).start(); }
  • /** -- **/
    • @RequestMapping(value="/brpush") @UnSession @ResponseBody publicboolean brpush(String key,String name,int age)throwsException{ boolean flag = redisService.rpush(key,JSON.toJSONString(newEdu(name,age))); return flag; }
  • /** **/
    • class blpopMessageHandler implementsRunnable{
  • privateString key;
  • privateRedisTemplate redisTemplate=(RedisTemplate)SpringContextUtil.getBean("redisTemplate");
  • publicString getKey(){
    • return key; }
  • publicvoid setKey(String key){
    • this.key = key; }
  • public blpopMessageHandler(String key){
    • this.key = key; }
  • @Override
    • publicvoid run(){
  • do{
  • //
    • String result = redisTemplate.blpop(1000,key);
  • if(EmptyUtil.isNotEmpty(result)){
  • //do something
    • Edu edu = JSON.parseObject(result,Edu.class);
  • System.out.println("【 】"+edu.getName()+"---->"+edu.getAge());
    • }
  • }while(true);
  • }
    • }
  • }

    • model

    1.  
    2. publicclassEduimplementsSerializable{
    3. privatestaticfinallong serialVersionUID =-6336530413316596246L;
    4. privateString name;
    5. privateint age;
    6. publicString getName(){
      7. return name; }
    7. publicvoid setName(String name){
      8. this.name = name; }
    8. publicint getAge(){
      9. return age; }
    9. publicvoid setAge(int age){
      10. this.age = age; }
    10. publicEdu(String name,int age){
      11. this.name = name; this.age = age; }
    11. @Override
      12. publicString toString(){ return"Edu{"+ "name='"+ name +'\''+ ", age="+ age + '}'; } }

      redis

      1.  
        2. /** * Jedis ， */ @Component publicclassRedisTemplate{ privatestaticLogger logger =LoggerFactory.getLogger(RedisTemplate.class);
      2. // pool jedis
        3. privatePool<Jedis> jedisPool;
      3. publicRedisTemplate(Pool<Jedis> jedisPool){
        4. this.jedisPool = jedisPool; }
      4. /**
        5. * REDIS * @param key reids * @param value */ publicboolean rpush(String key,String value){
      5. Jedis jedis =null;
      6. boolean flag =false;
        7. try{ jedis = jedisPool.getResource(); jedis.rpush(key.getBytes("utf-8"),value.getBytes("utf-8")); flag =true; }catch(Exception e){ // redis jedisPool.returnBrokenResource(jedis); e.printStackTrace(); }finally{ // close(jedis); }
      7. return flag;
        8. }
      8. /**
        9. * * @param key * @return */ publicString lpop(String key){
      9. byte[] bytes =null;
        10. Jedis jedis =null; try{
      10. jedis = jedisPool.getResource();
        11. bytes = jedis.lpop(key.getBytes("utf-8"));
      11. }catch(Exception e){
      12. // redis
        13. jedisPool.returnBrokenResource(jedis); e.printStackTrace();
      13. }finally{
      14. //
        15. close(jedis);
      15. }
        16. if(EmptyUtil.isNotEmpty(bytes)){ try{ returnnewString(bytes,"utf-8"); }catch(UnsupportedEncodingException e){ e.printStackTrace(); } }
      16. returnnull;
        17. }
      18. /**
        19. * list * @param timeout * @param key * @return */ publicString blpop(int timeout,String key){
      19. List<String> list =newArrayList<>();
      20. Jedis jedis =null;
      21. try{
      22. jedis = jedisPool.getResource();
      23. list = jedis.blpop(timeout,key);
      25. }catch(Exception e){
      26. // redis
        27. jedisPool.returnBrokenResource(jedis); e.printStackTrace();
      27. }finally{
      28. //
        29. close(jedis);
      29. }
        30. returnEmptyUtil.isNotEmpty(list)&& list.size()>1? list.get(1):null; }
      30. privatevoid close(Jedis jedis){
        31. try{ jedisPool.returnResource(jedis); }catch(Exception e){ if(jedis.isConnected()){ jedis.quit(); jedis.disconnect(); } } }
      31. }
    Redisベースのlocking実装
    Redisに必要な構成