Python异步Redis客户端与通用缓存装饰器

news2024/10/6 1:42:43

前言

这里我将通过 redis-py 简易封装一个异步的Redis客户端,然后主要讲解设计一个支持各种缓存代理(本地内存、Redis等)的缓存装饰器,用于在减少一些不必要的计算、存储层的查询、网络IO等。

具体代码都封装在 HuiDBK/py-tools: 打造 Python 开发常用的工具,让Coding变得更简单 (github.com) 中,以大家便捷使用。

异步redis客户端

首先安装 redis-py 库

pip install redis

Redis 之前是不支持异步的,后面为了统一异步redis操作与python常用的redis.py 的api接口一致,aioredis的作者已经将 aioredis 加入了redis中维护,安装的版本大于 4.2.0rc1 就行。

  • aioredis:https://github.com/aio-libs-abandoned/aioredis-py
  • redis:https://github.com/redis/redis-py

BaseRedisManager 封装

#!/usr/bin/python3
# -*- coding: utf-8 -*-
# @Author: Hui
# @Desc: { redis连接处理模块 }
# @Date: 2023/05/03 21:13
from datetime import timedelta
from typing import Optional, Union

from redis import Redis
from redis import asyncio as aioredis

from py_tools import constants
from py_tools.decorators.cache import CacheMeta, cache_json, RedisCacheProxy, AsyncRedisCacheProxy


class BaseRedisManager:
    """Redis客户端管理器"""

client: Union[Redis, aioredis.Redis] = None
    cache_key_prefix = constants.CACHE_KEY_PREFIX

    @classmethod
    def init_redis_client(
            cls,
            async_client: bool = False,
            host: str = "localhost",
            port: int = 6379,
            db: int = 0,
            password: Optional[str] = None,
            max_connections: Optional[int] = None,
            **kwargs
    ):
        """
        初始化 Redis 客户端。

        Args:
        async_client (bool): 是否使用异步客户端,默认为 False(同步客户端)
        host (str): Redis 服务器的主机名,默认为 'localhost'
        port (int): Redis 服务器的端口,默认为 6379
        db (int): 要连接的数据库编号,默认为 0
        password (Optional[str]): 密码可选
        max_connections (Optional[int]): 最大连接数。默认为 None(不限制连接数)
        **kwargs: 传递给 Redis 客户端的其他参数

        Returns:
        None
        """
        if cls.client is None:
            redis_client_cls = Redis
            if async_client:
                redis_client_cls = aioredis.Redis

            cls.client = redis_client_cls(
                host=host, port=port, db=db, password=password, max_connections=max_connections, **kwargs
            )

        return cls.client

    @classmethod
    def cache_json(
            cls,
            ttl: Union[int, timedelta] = 60,
            key_prefix: str = None,
    ):
        """
        缓存装饰器(仅支持缓存能够json序列化的数据)
        缓存函数整体结果
        Args:
        ttl: 过期时间 默认60s
        key_prefix: 默认的key前缀, 再未指定key时使用

        Returns:
        """
        key_prefix = key_prefix or cls.cache_key_prefix
        if isinstance(ttl, timedelta):
            ttl = int(ttl.total_seconds())

        cache_proxy = RedisCacheProxy(cls.client)
        if isinstance(cls.client, aioredis.Redis):
            cache_proxy = AsyncRedisCacheProxy(cls.client)

        return cache_json(cache_proxy=cache_proxy, key_prefix=key_prefix, ttl=ttl)

还是跟之前封装客户端一样的简易封装,由类属性 client 维护真正操作的redis的客户端,通过 init_redis_client 方法进行初始化。这样封装的目的就是在系统中只初始化一份 redis 客户端,操作时可以直接使用类方法。BaseRedisManager 只实现一些通用的 redis 操作(有待挖掘),具体还是需要业务Manager来继承封装业务中操作redis的方法。目前只实现了一个redis的缓存装饰器,其实内部就是组织参数设置redis代理,然后调用另外一个通用的缓存装饰器,这样使用的时候不需要制定缓存代理了。

缓存装饰器

#!/usr/bin/python3
# -*- coding: utf-8 -*-
# @Author: Hui
# @Desc: { 缓存装饰器模块 }
# @Date: 2023/05/03 19:23
import asyncio
import functools
import hashlib
import json
from datetime import timedelta

import cacheout
from redis import Redis
from redis import asyncio as aioredis
from pydantic import BaseModel, Field
from typing import Union
from py_tools import constants

MEMORY_PROXY = MemoryCacheProxy(cache_client=cacheout.Cache(maxsize=1024))


def cache_json(
        cache_proxy: BaseCacheProxy = MEMORY_PROXY,
        key_prefix: str = constants.CACHE_KEY_PREFIX,
        ttl: Union[int, timedelta] = 60,
):
    """
    缓存装饰器(仅支持缓存能够json序列化的数据)
    Args:
    cache_proxy: 缓存代理客户端, 默认系统内存
    ttl: 过期时间 默认60s
    key_prefix: 默认的key前缀

    Returns:
    """
    key_prefix = f"{key_prefix}:cache_json"
    if isinstance(ttl, timedelta):
        ttl = int(ttl.total_seconds())

    def _cache(func):

        def _gen_key(*args, **kwargs):
            """生成缓存的key"""

            # 根据函数信息与参数生成
            # key => 函数所在模块:函数名:函数位置参数:函数关键字参数 进行hash
            param_args_str = ",".join([str(arg) for arg in args])
            param_kwargs_str = ",".join(sorted([f"{k}:{v}" for k, v in kwargs.items()]))
            hash_str = f"{func.__module__}:{func.__name__}:{param_args_str}:{param_kwargs_str}"
            hash_ret = hashlib.sha256(hash_str.encode()).hexdigest()

            # 根据哈希结果生成key 默认前缀:函数所在模块:函数名:hash
            hash_key = f"{key_prefix}:{func.__module__}:{func.__name__}:{hash_ret}"
            return hash_key

        @functools.wraps(func)
        def sync_wrapper(*args, **kwargs):
            """同步处理"""

            # 生成缓存的key
            hash_key = _gen_key(*args, **kwargs)

            # 先从缓存获取数据
            cache_data = cache_proxy.get(hash_key)
            if cache_data:
                # 有直接返回
                print(f"命中缓存: {hash_key}")
                return json.loads(cache_data)

            # 没有,执行函数获取结果
            ret = func(*args, **kwargs)

            # 缓存结果
            cache_proxy.set(key=hash_key, value=json.dumps(ret), ttl=ttl)
            return ret

        @functools.wraps(func)
        async def async_wrapper(*args, **kwargs):
            """异步处理"""

            # 生成缓存的key
            hash_key = _gen_key(*args, **kwargs)

            # 先从缓存获取数据
            cache_data = await cache_proxy.get(hash_key)
            if cache_data:
                # 有直接返回
                return json.loads(cache_data)

            # 没有,执行函数获取结果
            ret = await func(*args, **kwargs)

            # 缓存结果
            await cache_proxy.set(key=hash_key, value=json.dumps(ret), ttl=ttl)
            return ret

        return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper

    return _cache

cache_json 是一个带单参数的缓存装饰器,可以指定一些缓存的代理、缓存key前缀、缓存ttl等。

内部实现了同步、异步函数的缓存处理,关键点其实就是如何构造唯一的缓存key,这里就是根据key前缀与函数的一些签名信息来构造的。

def _gen_key(*args, **kwargs):
    """生成缓存的key"""

    # 没有传递key信息,根据函数信息与参数生成
    # key => 函数所在模块:函数名:函数位置参数:函数关键字参数 进行hash
    param_args_str = ",".join([str(arg) for arg in args])
    param_kwargs_str = ",".join(sorted([f"{k}:{v}" for k, v in kwargs.items()]))
    hash_str = f"{func.__module__}:{func.__name__}:{param_args_str}:{param_kwargs_str}"
    hash_ret = hashlib.sha256(hash_str.encode()).hexdigest()

    # 根据哈希结果生成key 默认前缀:函数所在模块:函数名:hash
    hash_key = f"{key_prefix}:{func.__module__}:{func.__name__}:{hash_ret}"
    return hash_key

函数所在模块:函数名:函数位置参数:函数关键字参数 进行hash,在处理关键字参数的需要排个序,来保证相同的参数,顺序不同但缓存key一致。后面的逻辑就是常见的设置缓存操作。

image.png

缓存代理类

#!/usr/bin/python3
# -*- coding: utf-8 -*-
# @Author: Hui
# @Desc: { 缓存装饰器模块 }
# @Date: 2023/05/03 19:23
import asyncio
import functools
import hashlib
import json
from datetime import timedelta

import cacheout
from redis import Redis
from redis import asyncio as aioredis
from pydantic import BaseModel, Field
from typing import Union
from py_tools import constants


class CacheMeta(BaseModel):
    """缓存元信息"""

    key: str = Field(description="缓存的key")
    ttl: Union[int, timedelta] = Field(description="缓存有效期")
    cache_client: str = Field(description="缓存的客户端(Redis、Memcached等)")
    data_type: str = Field(description="缓存的数据类型(str、list、hash、set)")


class BaseCacheProxy(object):
    """缓存代理基类"""

    def __init__(self, cache_client):
        self.cache_client = cache_client  # 具体的缓存客户端,例如Redis、Memcached等

    def set(self, key: str, value: str, ttl: int):
        raise NotImplemented

    def get(self, key):
        cache_data = self.cache_client.get(key)
        return cache_data


class RedisCacheProxy(BaseCacheProxy):
    """同步redis缓存代理"""

    def __init__(self, cache_client: Redis):
        super().__init__(cache_client)

    def set(self, key, value, ttl):
        self.cache_client.setex(name=key, value=value, time=ttl)


class AsyncRedisCacheProxy(BaseCacheProxy):
    """异步Redis缓存代理"""

    def __init__(self, cache_client: aioredis.Redis):
        super().__init__(cache_client)

    async def set(self, key, value, ttl):
        await self.cache_client.setex(name=key, value=value, time=ttl)

    async def get(self, key):
        cache_data = await self.cache_client.get(key)
        return cache_data


class MemoryCacheProxy(BaseCacheProxy):
    """系统内存缓存代理"""

    def __init__(self, cache_client: cacheout.Cache):
        super().__init__(cache_client)

    def set(self, key, value, ttl):
        self.cache_client.set(key=key, value=value, ttl=ttl)


MEMORY_PROXY = MemoryCacheProxy(cache_client=cacheout.Cache(maxsize=1024))

这里设置一个缓存代理抽象类是用于封装屏蔽不同缓存客户端的操作不一致性。统一成如下入口

def set(self, key: str, value: str, ttl: int):
    raise NotImplemented

def get(self, key):
    cache_data = self.cache_client.get(key)
    return cache_data

让具体的缓存客户端重写(实现)这两个方法,以达到缓存装饰器的通用性。目前只实现了同步、异步redis缓存代理以及通过 cacheout 库实现的本地内存缓存代理,后面接入其他的缓存代理(例如Memcached等)就不用动cache_json函数了,只要继承 BaseCacheProxy,实现具体的 set、get 操作即可。

pip install python-memcached
import memcache

class MemcacheCacheProxy(BaseCacheProxy):

    def __init__(self, cache_client: memcache.Client):
        super().__init__(cache_client)

    def set(self, key, value, ttl):
        self.cache_client.set(key, value, time=ttl)

由于获取缓存的方法逻辑一致,故而直接复用就行。

测试Demo

#!/usr/bin/python3
# -*- coding: utf-8 -*-
# @Author: Hui
# @File: cache.py
# @Desc: { cache demo 模块 }
# @Date: 2024/04/23 11:11
import asyncio
import time
from datetime import timedelta

import cacheout

from py_tools.connections.db.redis_client import BaseRedisManager
from py_tools.decorators.cache import cache_json, MemoryCacheProxy, RedisCacheProxy, AsyncRedisCacheProxy


class RedisManager(BaseRedisManager):
    client = None


class AsyncRedisManager(BaseRedisManager):
    client = None


RedisManager.init_redis_client(async_client=False)
AsyncRedisManager.init_redis_client(async_client=True)

memory_proxy = MemoryCacheProxy(cache_client=cacheout.Cache())
redis_proxy = RedisCacheProxy(cache_client=RedisManager.client)
aredis_proxy = AsyncRedisCacheProxy(cache_client=AsyncRedisManager.client)


@cache_json(key_prefix="demo", ttl=3)
def memory_cache_demo_func(name: str, age: int):
    return {"test_memory_cache": "hui-test", "name": name, "age": age}


@cache_json(cache_proxy=redis_proxy, ttl=10)
def redis_cache_demo_func(name: str, age: int):
    return {"test_redis_cache": "hui-test", "name": name, "age": age}


@cache_json(cache_proxy=aredis_proxy, ttl=timedelta(minutes=1))
async def aredis_cache_demo_func(name: str, age: int):
    return {"test_async_redis_cache": "hui-test", "name": name, "age": age}


@AsyncRedisManager.cache_json(ttl=30)
async def aredis_manager_cache_demo_func(name: str, age: int):
    return {"test_async_redis_manager_cache": "hui-test", "name": name, "age": age}


def memory_cache_demo():
    print("memory_cache_demo")
    ret1 = memory_cache_demo_func(name="hui", age=18)
    print("ret1", ret1)
    print()

    ret2 = memory_cache_demo_func(name="hui", age=18)
    print("ret2", ret2)
    print()

    time.sleep(3)
    ret3 = memory_cache_demo_func(age=18, name="hui")
    print("ret3", ret3)
    print()

    assert ret1 == ret2 == ret3


def redis_cache_demo():
    print("redis_cache_demo")
    ret1 = redis_cache_demo_func(name="hui", age=18)
    print("ret1", ret1)
    print()

    ret2 = redis_cache_demo_func(name="hui", age=18)
    print("ret2", ret2)

    assert ret1 == ret2


async def aredis_cache_demo():
    print("aredis_cache_demo")
    ret1 = await aredis_cache_demo_func(name="hui", age=18)
    print("ret1", ret1)
    print()

    ret2 = await aredis_cache_demo_func(name="hui", age=18)
    print("ret2", ret2)

    assert ret1 == ret2


async def aredis_manager_cache_demo():
    print("aredis_manager_cache_demo")
    ret1 = await aredis_manager_cache_demo_func(name="hui", age=18)
    print("ret1", ret1)
    print()

    ret2 = await aredis_manager_cache_demo_func(name="hui", age=18)
    print("ret2", ret2)

    assert ret1 == ret2


async def main():
    memory_cache_demo()

    redis_cache_demo()

    await aredis_cache_demo()

    await aredis_manager_cache_demo()


if __name__ == '__main__':
    asyncio.run(main())

输出结果

memory_cache_demo
ret1 {'test_memory_cache': 'hui-test', 'name': 'hui', 'age': 18}

命中缓存: demo:cache_json:__main__:memory_cache_demo_func:46c6a618a88eb5067a00915c10c97c6c72d5073ecf9b04060433de75b2d21f51
ret2 {'test_memory_cache': 'hui-test', 'name': 'hui', 'age': 18}

ret3 {'test_memory_cache': 'hui-test', 'name': 'hui', 'age': 18}

redis_cache_demo
ret1 {'test_redis_cache': 'hui-test', 'name': 'hui', 'age': 18}

命中缓存: py-tools:cache_json:__main__:redis_cache_demo_func:a00b13aa2e1e56ad328d1956bc3c3fb8e89b7007453a780e866cc3ccafb51d73
ret2 {'test_redis_cache': 'hui-test', 'name': 'hui', 'age': 18}
aredis_cache_demo
ret1 {'test_async_redis_cache': 'hui-test', 'name': 'hui', 'age': 18}

ret2 {'test_async_redis_cache': 'hui-test', 'name': 'hui', 'age': 18}
aredis_manager_cache_demo
ret1 {'test_async_redis_manager_cache': 'hui-test', 'name': 'hui', 'age': 18}

ret2 {'test_async_redis_manager_cache': 'hui-test', 'name': 'hui', 'age': 18}

Redis 缓存情况

缓存信息还是挺清晰的就是有点长。由于是从主入口调用的函数,所以 func.__module__ 是 __main__。

这缓存装饰器一般适用于一些参数相同, 结果经常不变的情况下,以及允许短时间内出现数据不一致的场景。如下是一些典型的应用场景

  1. API Token缓存:对于需要使用API Token进行身份验证的API请求,通常API Token具有一定的有效期。在这种情况下,你可以使用缓存装饰器来缓存API Token,以避免在每次请求时重新生成或从后端服务获取。这样可以降低对后端服务的负载,并提高系统的响应速度。
  2. OSS Sign URL缓存:当需要生成签名URL来访问对象存储服务(如AWS S3、阿里云OSS)中的资源时,通常需要对URL进行签名以确保安全性。在这种情况下,你可以使用缓存装饰器来缓存已签名的URL,在一定时间内重复使用相同的签名URL,而不必重新计算签名。这样可以降低对签名计算资源的消耗,并减少重复的签名请求。
  3. 频繁查询的数据缓存:对于一些数据不经常变化但是频繁被查询的情况,比如一些静态配置信息、全局参数等,可以使用缓存装饰器将查询结果缓存起来,减少数据库查询次数,提高系统的响应速度。
  4. 外部API响应结果缓存:当你调用外部API获取数据时,有时这些数据在一段时间内不会发生变化。在这种情况下,你可以使用缓存装饰器来缓存外部API的响应结果,以避免频繁地向外部API发出请求。这不仅可以提高系统的性能,还可以降低对外部服务的依赖性。

总的来说,缓存装饰器可以应用于许多场景,特别是在需要提高性能、减少资源消耗和避免重复请求数据的情况下。通过合理地设置缓存时间,可以权衡数据的新鲜度和系统性能,从而实现更好的用户体验。

源代码

源代码已上传到了Github,里面也有具体的使用Demo,欢迎大家一起体验、贡献。

HuiDBK/py-tools: 打造 Python 开发常用的工具,让Coding变得更简单 (github.com)

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/1638763.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

使用 uni-app 开发 iOS 应用的操作步骤

哈喽呀,大家好呀,淼淼又来和大家见面啦,上一期和大家一起探讨了使用uniapp开发iOS应用的优势及劣势之后有许多小伙伴想要尝试使用uniapp开发iOS应用,但是却不懂如何使用uniapp开发iOS应用,所以这一期淼淼就来给你们分享…

nginx--平滑升级

失败了,等我拍好错继续更新 命令 选项说明 帮助: -? -h 使用指定的配置文件: -c 指定配置指令:-g 指定运行目录:-p 测试配置文件是否有语法错误:-t -T 打印nginx的版本信息、编译信息等:-v -V 发送信号: -s 示例: nginx -s reload 信号说明 立刻停止服务:stop,相…

【C++】学习笔记——string_3

文章目录 六、string类5. string类的操作6. string类的转换7. string类的模拟实现 未完待续 搭配文档食用 六、string类 5. string类的操作 上面的函数中,有些是不常用的,咱们只挑几个重要的进行讲解。 c_str 就是将字符串转换成 C语言 字符串的格式。…

[Java EE] 多线程(六):线程池与定时器

1. 线程池 1.1 什么是线程池 我们前面提到,线程的创建要比进程开销小,但是如果线程的创建/销毁比较频繁,开销也会比较大.所以我们便引入了线程池,线程池的作用就是提前把线程都创建好,放到用户态代码中写的数据结构中,后面就可以随用随取. 线程池最大的好处就是减少每次启动,…

Python中动画显示与gif生成

1. 动画生成 主要使用的是 matplotlib.animation ,具体示例如下: import matplotlib.pyplot as plt import matplotlib.animation as animation import numpy as np fig, ax plt.subplots() t np.linspace(0, 3, 40) g -9.81 v0 12 z g * t**2 / …

【Python函数和类6/6】类与对象

目录 目标 类与对象 类的定义 栗子 实例化对象 属性和方法的调用 特殊的self参数 类方法的其它参数 函数与方法的区别 总结 目标 在前面的博客当中,我们已经接触了一部分封装。比如:将数据扔进列表中,这就是一个简单…

短视频素材去哪里搬运?短视频素材有哪些类型?

在这个数字化和视觉传达至关重要的时代,选择合适的视频素材对于提升视频内容的吸引力和观众参与度至关重要。无论您是一名广告制片人、社交媒体经理还是独立视频制作者,以下这些精选的视频素材网站将为您提供从高清视频到特效资源的全面支持,…

工厂模式和策略模式区别

工厂模式和策略模式都是面向对象设计模式,但它们的目的和应用场景有所不同。 工厂模式是一种创建型设计模式,旨在通过使用一个工厂类来创建对象,而不是直接使用new关键字来创建对象。这样做可以使系统更容易扩展和维护,因为新的对…

reactjs后台管理系统搭建

1 通过yarn 模板创建reactjs项目 yarn create vite reactjs-antdesign-admin --template react-ts 2 基础路由测试 定义一个router/index.tsx&#xff0c;里面定义路由组件 const Router: React.FC () > {return (<HashRouter><Switch><Route path"…

Edge浏览器使用心得与深度探索

&#x1f90d; 前端开发工程师、技术日更博主、已过CET6 &#x1f368; 阿珊和她的猫_CSDN博客专家、23年度博客之星前端领域TOP1 &#x1f560; 牛客高级专题作者、打造专栏《前端面试必备》 、《2024面试高频手撕题》 &#x1f35a; 蓝桥云课签约作者、上架课程《Vue.js 和 E…

信息时代的智慧导航:高效搜索、信息筛选与信任构建的全面指南!

文章目录 一、高效搜索&#xff1a;快速定位目标信息的秘诀二、信息筛选&#xff1a;去伪存真&#xff0c;找到有价值的信息三、信任构建&#xff1a;深入了解与直接沟通《搜索之道&#xff1a;信息素养与终身学习的新引擎》亮点内容简介目录获取方式 随着科技的飞速发展&#…

前端基础学习html(2)

目录 表格标签&#xff1a; 列表标签&#xff1a; 表格标签&#xff1a; <!-- 表格基本架构 --><!-- tr表示一行&#xff0c;td表示一行内单元格 --><!--th为第一行表头加粗居中显示 --><table border"1"><thead><tr><th&g…

用Stream流方式合并两个list集合(部分对象属性重合)

一、合并出共有部分 package com.xu.demo.test;import java.util.Arrays; import java.util.List; import java.util.stream.Collectors;public class ListMergeTest1 {public static void main(String[] args) {List<User> list1 Arrays.asList(new User(1, "Alic…

【linux学习指南】linux指令与实践文件编写

文章目录 &#x1f4dd;前言&#x1f320; linux目录结构&#x1f309;linux命令介绍 &#x1f320;pwd命令&#x1f309;mkdir指令&#xff08;重要&#xff09; &#x1f320;cd 指令&#x1f309;touch指令 &#x1f320;rmdir指令 && rm 指令&#xff08;重要&…

nginx--配置文件

组成 主配置文件&#xff1a;nginx.conf 子配置文件&#xff1a;include conf.d/*.conf 协议相关的配置文件&#xff1a;fastcgi uwsgi scgi等 mime.types&#xff1a;⽀持的mime类型&#xff0c;MIME(Multipurpose Internet Mail Extensions)多用途互联⽹网邮件扩展类型&…

KUKA机器人KR3 R540维护保养——涂润滑脂

KUKA机器人在保养时少不了润滑脂&#xff0c;不同型号的机器人需要的润滑脂类型也不一样&#xff0c;保养时注意选用合适的润滑脂。本篇文章以KUKA机器人KR3 R540为例&#xff0c;在轴盖板 A2、A3、A5 的内侧涂上润滑脂。 一、涂润滑脂的作用 拆开机器人一个轴的盖板&am…

链表面试题2

1&#xff0c;合并两个有序链表 我们先定义一个虚拟节点newH&#xff0c; 然后按照上图所走&#xff0c;但是当其中一个链表走空时&#xff0c;我们只需返回另一个链表即可 class Solution {public ListNode mergeTwoLists(ListNode headA, ListNode headB) {ListNode newhead…

python基础语法--函数

一、函数概述 函数就是执行特定任务完成特定功能的一段代码。可以在程序中将某一段代码定义成函数&#xff0c;并指定一个函数名和接收的输入&#xff08;参数&#xff09;&#xff0c;这样就可以在程序的其他地方通过函数名多次调用并执行该段代码了。 每次调用执行后&#…

Mybatis-Plus学习:快速入门、核心功能、扩展功能、插件功能

文章目录 MybatisPlus快速入门快速开始常见注解常见配置 核心功能条件构造器&#xff08;Wrapper&#xff09;自定义SQLService接口基本用法基础业务接口复杂业务接口Lamda查询Lamda更新批量新增 扩展功能代码生成代码生成器快速开发插件 静态工具逻辑删除枚举处理器JSON处理器…

机器人系统ros2-开发实践04-ROS 2 启动文件管理大型项目的最佳实践

机器人上的大型应用通常涉及多个互连的节点&#xff0c;每个节点可以有许多参数。海龟模拟器中模拟多只海龟就是一个很好的例子。海龟模拟由多个海龟节点、世界配置以及 TF 广播器和监听器节点组成。在所有节点之间&#xff0c;存在大量影响这些节点的行为和外观的 ROS 参数。 …