码农知识堂 - 1000bd
  •   Python
  •   PHP
  •   JS/TS
  •   JAVA
  •   C/C++
  •   C#
  •   GO
  •   Kotlin
  •   Swift

  • lettuce利用stream实现消息推送

    1、消息推送

    1. /**
    2.  * @Auther: pshdhx
    3.  * @Date: 2023/02/22/10:38
    4.  * @Description: 往同一个stream队列里边塞值，同一队列的所有消费者组，都会收到消息
    5.  * 模拟 消息推送到服务器
    6.  */
    7. public class TestPubStream {
    8.     public static void main(String[] args) {
    9.         // 创建 Redis 连接
    10.         RedisURI redisURI = RedisURI.Builder.redis("xxxxxx", 6379).build();
    11.         redisURI.setPassword("xxxxxx?");
    12.         RedisClient redisClient = RedisClient.create(redisURI);
    13.         StatefulRedisConnection connection = redisClient.connect();
    14.         try {
    15.             // 获取同步命令对象
    16.             RedisCommands syncCommands = connection.sync();
    17.             // 创建消费者组
    18.             String streamKey = "mystream"; // Stream名称
    19.             String groupName = "myConsumer"; // 消费者组名称
    20.             //检查 groupName 是否存在
    21.             boolean groupExists = false;
    22.             List result = syncCommands.xinfoGroups(streamKey);
    23.             for (Object obj : result) {
    24.                 ArrayList objList = (java.util.ArrayList) obj;
    25.                 if((objList.get(1)+"").equals(groupName)){
    26.                     groupExists = true;
    27.                     break;
    28.                 }
    29.             }
    30.  
    31.             System.out.println("groupExists = " + groupExists);
    32.  
    33.             // 如果 groupName 不存在，则创建
    34.             if (!groupExists) {
    35.                 syncCommands.xgroupCreate(XReadArgs.StreamOffset.from(streamKey, "0-0"), groupName, XGroupCreateArgs.Builder.mkstream());
    36.             }
    37.             // 发布消息
    38.             syncCommands.xadd(streamKey, "key1", "val1");
    39.  
    40.         } finally {
    41.             // 关闭连接
    42.             connection.close();
    43.             redisClient.shutdown();
    44.         }
    45.     }
    46. }
      47.  

      2、实现消息订阅
       

      1. /**
      2.  * @Auther: pshdhx
      3.  * @Date: 2023/02/22/10:40
      4.  * @Description: 模拟 tomcat1 使用消费者组1，消费者1向redis 服务器 订阅推送的消息
      5.  */
      6.  
      7. public class TestSub1_tomcat1 {
      8.     public static void main(String[] args) {
      9.         // 创建 Redis 连接
      10.         RedisURI redisURI = RedisURI.Builder.redis("xxxxxx", 6379).build();
      11.         redisURI.setPassword("xxxxxxxxxx");
      12.         RedisClient redisClient = RedisClient.create(redisURI);
      13.         StatefulRedisConnection connection = redisClient.connect();
      14.         try {
      15.           
      16.  
      17.             // 从消费者组中获取消息
      18.             String consumerName = "tomcat1_consumer_name"; // 消费者名称
      19.             RedisCommands streamCommands = connection.sync();
      20.  
      21.             while (true) {
      22.                 List> messages = streamCommands.xreadgroup(
      23.                         Consumer.from(groupName, consumerName),
      24.                         XReadArgs.Builder.block(Duration.ofSeconds(5)), // 阻塞5秒钟等待新消息
      25.                         XReadArgs.StreamOffset.lastConsumed(streamKey)
      26.                 );
      27.  
      28.                 if (messages.isEmpty()) {
      29.                     continue; // 在没有新消息时继续轮询
      30.                 }
      31.  
      32.                 for (StreamMessage message : messages) {
      33.                     System.out.println("tomcat 1 Received message: " + message.getBody());
      34.  
      35.                     // 手动确认消息已被处理
      36.                     syncCommands.xack(streamKey, groupName, message.getId());
      37.                 }
      38.             }
      39.  
      40.         } finally {
      41.             // 关闭连接
      42.             connection.close();
      43.             redisClient.shutdown();
      44.         }
      45.     }
      46. }
       

      
                
                    
                

                
    47. 

                    相关阅读:
      
                    
一体化运维监控：提供全方位数据洞察                            
      
项目中常用工具包                            
      
配置文件自动提示                            
      
Redis(四) 主从、哨兵、集群环境搭建                            
      
ERP系统中有采购管理模块，为什么企业要选择供应商管理SRM系统？                            
      
基于FFMPEG+SDL的简单的视频播放器分析                            
      
ESP32 Arduino实战协议篇-BLE 客户端实现温度和湿度数据传输                            
      
快速MOCK数据并插入数据表中（MySQL）                            
      
网络安全（黑客）自学                            
      
Stack Overflow开发者调查发布：AI将如何协助DevOps                            
      
                          
                
      48. 
                
    48. 
                    原文地址：https://blog.csdn.net/pshdhx/article/details/133179644
                
      49. 

            
        

        
      
            
        
                
      • 
                    最新文章
                
        • 
                
      • 
                    
攻防演习之三天拿下官网站群                            
        
数据安全治理学习——前期安全规划和安全管理体系建设                            
        
企业安全 | 企业内一次钓鱼演练准备过程                            
        
内网渗透测试 | Kerberos协议及其部分攻击手法                            
        
0day的产生 | 不懂代码的"代码审计"                            
        
安装scrcpy-client模块av模块异常，环境问题解决方案                            
        
leetcode hot100【LeetCode 279. 完全平方数】java实现                            
        
OpenWrt下安装Mosquitto                            
        
AnatoMask论文汇总                            
        
【AI日记】24.11.01 LangChain、openai api和github copilot                            
        
                            
                
        • 
            
      

            
        
                
      • 
                    热门文章
                
        • 
                
      • 
                    
十款代码表白小特效 一个比一个浪漫 赶紧收藏起来吧！！！                            
        
奉劝各位学弟学妹们，该打造你的技术影响力了！                            
        
五年了，我在 CSDN 的两个一百万。                            
        
Java俄罗斯方块，老程序员花了一个周末，连接中学年代！                            
        
面试官都震惊，你这网络基础可以啊！                            
        
你真的会用百度吗？我不信 — 那些不为人知的搜索引擎语法                            
        
心情不好的时候，用 Python 画棵樱花树送给自己吧                            
        
通宵一晚做出来的一款类似CS的第一人称射击游戏Demo！原来做游戏也不是很难，连憨憨学妹都学会了！                            
        
13 万字 C 语言从入门到精通保姆级教程2021 年版                            
        
10行代码集2000张美女图，Python爬虫120例，再上征途                            
        
                            
                
        • 
            
      

        
      
    




    
    
    
    

      
    
      
        
      
            
      
                
      
                    Copyright © 2022 侵权请联系2656653265@qq.com   
                    京ICP备2022015340号-1
                
      
                
      
                    正则表达式工具
                    cron表达式工具
                    密码生成工具
                
      
                
      
                    
                    
      京公网安备 11010502049817号