redis 限流redis 令牌桶限流@dataclass(frozen=True) class RateLimitResult: allowed: bool remaining_tokens: float retry_after_seconds: float now_ms: int class RedisTokenBucketRateLimiter: _TOKEN_BUCKET_LUA = r""" local key = KEYS[1] -- 令牌桶状态存储的 Redis Key（Hash） local capacity = tonumber(ARGV[1]) -- 桶容量：最多可累积的令牌数 local rate = tonumber(ARGV[2]) -- 补充速率：每秒补充多少令牌（tokens/s） local requested = tonumber(ARGV[3]) -- 本次请求要消耗的令牌数 local ttl_ms = tonumber(ARGV[4]) -- 令牌桶 Key 的过期时间（毫秒），用于回收冷 Key local t = redis.call('TIME') -- 读取 Redis 服务器时间：{秒, 微秒} local now = tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000) -- 服务器时间换算为毫秒时间戳 local data = redis.call('HMGET', key, 'tokens', 'ts') -- 读取当前令牌数(tokens)与上次更新时间(ts) local tokens = tonumber(data[1]) -- 当前令牌数（可能为 nil） local ts = tonumber(data[2]) -- 上次更新时间戳（毫秒，可能为 nil） if tokens == nil then tokens = capacity end -- 初始化：若不存在，则认为桶是满的 if ts == nil then ts = now end -- 初始化：若不存在，则将上次更新时间设为当前 if now < ts then ts = now end -- 防御：若时钟回拨导致 now < ts，则强制对齐 local delta_ms = now - ts -- 距离上次更新过去了多少毫秒 local refill = (delta_ms / 1000.0) * rate -- 这段时间应补充的令牌数量（可为小数） tokens = math.min(capacity, tokens + refill) -- 将令牌补充后并截断到容量上限 local allowed = 0 -- 是否允许：0/1（Lua 没有布尔返回给 Python 的统一类型） local retry_after_ms = 0 -- 如果不允许，建议等待多少毫秒再试 if tokens >= requested then -- 若令牌充足 allowed = 1 -- 标记允许 tokens = tokens - requested -- 扣除本次消耗 else -- 否则令牌不足 allowed = 0 -- 标记拒绝 if rate > 0 then -- 若补充速率 > 0，可计算需要等待的时间 local missing = requested - tokens -- 还差多少令牌 retry_after_ms = math.ceil((missing / rate) * 1000) -- 差额按速率换算为等待毫秒数（向上取整） else -- 若 rate=0，则永远补不回令牌 retry_after_ms = -1 -- 用 -1 表示无法通过等待获得令牌 end -- 结束 rate>0 分支 end -- 结束 tokens>=requested 分支 redis.call('HMSET', key, 'tokens', tokens, 'ts', now) -- 写回最新 tokens 与 ts（原子更新） if ttl_ms ~= nil and ttl_ms > 0 then -- 若配置了过期时间且 >0 redis.call('PEXPIRE', key, ttl_ms) -- 给令牌桶 Key 设置过期（毫秒） end -- 结束 ttl 分支 return {allowed, tokens, retry_after_ms, now} -- 返回：是否允许、剩余令牌、建议等待(ms)、当前时间(ms) """ def __init__( self, redis: Optional[AsyncRedis] = None, *, capacity: Union[int, float], refill_rate: Union[int, float], requested: Union[int, float] = 1, ttl_ms: Optional[int] = None, key_prefix: str = "rate_limit:token_bucket:", ) -> None: capacity_f = float(capacity) rate_f = float(refill_rate) req_f = float(requested) if capacity_f <= 0: raise ValueError("capacity 必须 > 0") if rate_f < 0: raise ValueError("refill_rate 必须 >= 0") if req_f <= 0: raise ValueError("requested 必须 > 0") if req_f > capacity_f: raise ValueError("requested 不能大于 capacity") self._redis = redis or async_redis_client self._key_prefix = key_prefix self._capacity = capacity_f self._refill_rate = rate_f self._requested = req_f self._ttl_ms = int(ttl_ms) if ttl_ms is not None else None def _full_key(self, key: str) -> str: return f"{self._key_prefix}{key}" @staticmethod def _default_ttl_ms( capacity: Union[int, float], refill_rate: Union[int, float] ) -> int: # 让桶在“完全补满所需时间”的 2 倍后过期，避免大量冷 key 常驻 # 至少 5 秒，防止极小桶/极大 rate 造成频繁抖动 if refill_rate <= 0: return 60_000 seconds = max(5.0, float(capacity) / float(refill_rate) * 2.0) return int(seconds * 1000) async def allow( self, key: str, *, requested: Optional[Union[int, float]] = None, ttl_ms: Optional[int] = None, ) -> RateLimitResult: """ 尝试拿令牌（令牌桶）。 - capacity/refill_rate/requested 默认来自 __init__ - requested/ttl_ms 允许在单次调用中覆盖 """ capacity_f = self._capacity rate_f = self._refill_rate req_f = float(requested) if requested is not None else self._requested if req_f <= 0: raise ValueError("requested 必须 > 0") if req_f > capacity_f: raise ValueError("requested 不能大于 capacity") ttl_ms_i = ( int(ttl_ms) if ttl_ms is not None else ( self._ttl_ms if self._ttl_ms is not None else self._default_ttl_ms(capacity_f, rate_f) ) ) full_key = self._full_key(key) res = await self._redis.eval( self._TOKEN_BUCKET_LUA, 1, full_key, capacity_f, rate_f, req_f, ttl_ms_i, ) allowed = bool(int(res[0])) remaining = float(res[1]) retry_after_ms = int(res[2]) now_ms = int(res[3]) retry_after_seconds = 0.0 if not allowed: retry_after_seconds = ( 0.0 if retry_after_ms <= 0 else retry_after_ms / 1000.0 ) return RateLimitResult( allowed=allowed, remaining_tokens=remaining, retry_after_seconds=retry_after_seconds, now_ms=now_ms, ) async def wait( self, key: str, *, requested: Optional[Union[int, float]] = None, ttl_ms: Optional[int] = None, max_wait_seconds: Optional[float] = None, poll_min_sleep: float = 0.01, ) -> RateLimitResult: """ 一直等到拿到令牌为止（适合 Activity 里对外部依赖做限流）。 max_wait_seconds: 超过则抛 TimeoutError """ start = time.perf_counter() while True: r = await self.allow( key, requested=requested, ttl_ms=ttl_ms, ) if r.allowed: return r sleep_s = max(poll_min_sleep, r.retry_after_seconds) if max_wait_seconds is not None: elapsed = time.perf_counter() - start if elapsed + sleep_s > max_wait_seconds: raise TimeoutError( f"rate limit wait timeout: key={key}, waited={elapsed:.3f}s, next_sleep={sleep_s:.3f}s" ) await asyncio.sleep(sleep_s) redis 滑动窗口限流@dataclass(frozen=True) class SlidingWindowResult: allowed: bool current_count: int retry_after_seconds: float now_ms: int class RedisSlidingWindowRateLimiter: """ Redis 滑动窗口限流（ZSET）。 语义：任意连续 window_seconds 秒内最多 limit 次通过。 """ _SLIDING_WINDOW_LUA = r""" local key = KEYS[1] -- 滑动窗口限流使用的 ZSET Key local window_ms = tonumber(ARGV[1]) -- 窗口长度（毫秒） local limit = tonumber(ARGV[2]) -- 窗口内允许的最大通过次数 local ttl_ms = tonumber(ARGV[3]) -- Key 过期时间（毫秒），用于回收冷 Key -- Redis server time (ms) -- 注释：以下用 Redis TIME 作为统一时钟，避免多机时间不一致 local t = redis.call('TIME') -- 读取 Redis 服务器时间：{秒, 微秒} local now = tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000) -- 转为毫秒时间戳 -- window cleanup -- 注释：删除窗口之外的旧请求记录 local start = now - window_ms -- 窗口起点（毫秒） redis.call('ZREMRANGEBYSCORE', key, '-inf', start) -- 删除 score <= start 的成员（窗口外） local cnt = tonumber(redis.call('ZCARD', key)) -- 统计窗口内当前已有多少次请求 if cnt < limit then -- 若窗口内次数仍未达到上限（严格 <，避免多放 1 次） -- unique member to avoid overwrite under same ms -- 注释：member 唯一化，避免同毫秒覆盖导致少计数 local seq = redis.call('INCR', key .. ':seq') -- 递增序列号（辅助唯一 member） local member = tostring(now) .. ':' .. tostring(seq) -- member = now:seq redis.call('ZADD', key, now, member) -- 写入本次请求记录：score=now(member 的时间) if ttl_ms ~= nil and ttl_ms > 0 then -- 若配置了过期时间且 >0 redis.call('PEXPIRE', key, ttl_ms) -- 给 ZSET Key 设置过期时间 redis.call('PEXPIRE', key .. ':seq', ttl_ms) -- 给序列号 Key 也设置过期时间 end -- 结束 ttl 分支 return {1, cnt + 1, 0, now} -- 返回：允许(1)、窗口内计数(含本次)、建议等待(ms=0)、now(ms) else -- 否则窗口内次数已满，需要拒绝 -- compute retry_after_ms: when the oldest entry exits window -- 注释：计算何时最早记录滑出窗口 local oldest = redis.call('ZRANGE', key, 0, 0, 'WITHSCORES') -- 取窗口内最早的一条（score 最小） local oldest_ts = nil -- 最早记录的时间戳（毫秒） if oldest ~= nil and #oldest >= 2 then -- 若存在成员且带 score oldest_ts = tonumber(oldest[2]) -- oldest[2] 为 score（ZSET 的时间戳） end -- 结束 oldest 解析 local retry_after = 0 -- 建议等待的毫秒数 if oldest_ts ~= nil then -- 若能拿到最早时间戳 retry_after = math.max(0, math.ceil((oldest_ts + window_ms) - now)) -- 等到 oldest_ts+window_ms 才会滑出 else -- 极端情况：ZSET 为空但 cnt>=limit（理论不应发生） retry_after = math.ceil(window_ms) -- 保守等待一个窗口长度 end -- 结束 oldest_ts 分支 if ttl_ms ~= nil and ttl_ms > 0 then -- 若配置了过期时间且 >0 redis.call('PEXPIRE', key, ttl_ms) -- 维持 key 的过期时间（避免热 key 被误删） redis.call('PEXPIRE', key .. ':seq', ttl_ms) -- 维持 seq key 的过期时间 end -- 结束 ttl 分支 return {0, cnt, retry_after, now} -- 返回：拒绝(0)、当前窗口计数、建议等待(ms)、now(ms) end -- 结束 if cnt < limit 分支 """ def __init__( self, redis: Optional[AsyncRedis] = None, *, limit: int, window_seconds: float, ttl_ms: Optional[int] = None, key_prefix: str = "rate_limit:sliding_window:", ) -> None: if limit <= 0: raise ValueError("limit 必须 > 0") if window_seconds <= 0: raise ValueError("window_seconds 必须 > 0") self._redis = redis or async_redis_client self._key_prefix = key_prefix self._limit = int(limit) self._window_seconds = float(window_seconds) self._ttl_ms = int(ttl_ms) if ttl_ms is not None else None def _full_key(self, key: str) -> str: return f"{self._key_prefix}{key}" @staticmethod def _default_ttl_ms(window_seconds: float) -> int: # key 存活时间略大于窗口，方便清理且避免冷 key 常驻 return int(max(5.0, window_seconds * 2.0) * 1000) async def allow( self, key: str, *, ttl_ms: Optional[int] = None, ) -> SlidingWindowResult: window_ms = int(self._window_seconds * 1000) ttl_ms_i = ( int(ttl_ms) if ttl_ms is not None else ( self._ttl_ms if self._ttl_ms is not None else self._default_ttl_ms(self._window_seconds) ) ) res = await self._redis.eval( self._SLIDING_WINDOW_LUA, 1, self._full_key(key), window_ms, self._limit, ttl_ms_i, ) allowed = bool(int(res[0])) current_count = int(res[1]) retry_after_ms = int(res[2]) now_ms = int(res[3]) return SlidingWindowResult( allowed=allowed, current_count=current_count, retry_after_seconds=retry_after_ms / 1000.0 if retry_after_ms > 0 else 0.0, now_ms=now_ms, ) async def wait( self, key: str, *, ttl_ms: Optional[int] = None, max_wait_seconds: Optional[float] = None, poll_min_sleep: float = 0.01, ) -> SlidingWindowResult: start = asyncio.get_running_loop().time() while True: r = await self.allow( key, ttl_ms=ttl_ms, ) if r.allowed: return r sleep_s = max(poll_min_sleep, r.retry_after_seconds) if max_wait_seconds is not None: elapsed = asyncio.get_running_loop().time() - start if elapsed + sleep_s > max_wait_seconds: raise TimeoutError( f"sliding window wait timeout: key={key}, waited={elapsed:.3f}s, next_sleep={sleep_s:.3f}s" ) await asyncio.sleep(sleep_s)
user nobody; # Nginx worker进程运行的用户及用户组 worker_processes 4; # 在master/worker运行方式下，定义worker进程的个数; 可以为auto worker_rlimit_nofile 65535; # 设置 Nginx worker 进程可以打开的最大文件描述符数量。这个指令通常用于确保 Nginx 能够处理大量并发连接，因为每个连接通常会占用一个文件描述符。 worker_rlimit_core 1m; # 通过worker_rlimit_core配置可以限制core文件的大小 working_directory /usr/local/lighthouse/softwares/nginx/coredump; # 设置coredump文件所放置的目录 daemon on; # 是否以守护进程方式运行Nginx master_process on; # 是否以 master/worker 方式工作 worker_cpu_affinity 1000 0100 0010 0001; # 绑定worker进程到指定的CPU内核 error_log logs/error.log notice; # error_log/path/file level; timer_resolution 100ms; # 每次内核的事件调用（如epoll、select、poll、kqueue等）返回时，都会执行一次gettimeofday，实现用内核的时钟来更新Nginx中的缓存时钟。 events { # 处理网络连接设置的核心部分 debug_connection 10.224.66.14; # 仅仅来自以上IP地址的请求才会输出debug级别的日志 debug_connection 10.224.57.0/24; # 其他请求仍然沿用error_log中配置的日志级别 accept_mutex on; # 负载均衡锁 accept锁默认是打开的，如果关闭它，那么建立TCP连接的耗时会更短，但worker进程之间的负载会非常不均衡 accept_mutex_delay 500ms; # 使用accept锁后，同一时间只有一个worker进程能够取到accept锁。这个accept锁不是阻塞锁，如果取不到会立刻返回。 如果有一个worker进程试图取accept锁而没有取到，它至少要等accept_mutex_delay定义的时间间隔后才能再次试图取锁。 multi_accept off; # 当事件模型通知有新连接时，尽可能地对本次调度中客户端发起的所有TCP请求都建立连接。 use epoll; # 事件驱动模型 worker_connections 8192; # 每个worker进程可以同时处理的最大连接数。 } http { include mime.types; default_type application/octet-stream; log_format combinedio '$remote_addr - $remote_user [$time_local] ' '"$request" $status $body_bytes_sent ' '"$http_referer" "$http_user_agent" $request_length $request_time $upstream_response_time'; access_log logs/access.log combinedio; sendfile on; # 硬盘 >> kernel buffer >> user buffer>> kernel socket buffer >>协议栈 | 硬盘 >> kernel buffer (快速拷贝到kernelsocket buffer) >>协议栈 gzip on; # 压缩响应内容 tcp_nopush on; # 数据量达到一定值后发送 tcp_nodelay on; # 对keepalive连接是否使用TCP_NODELAY选项。# 应用程序每产生一次操作就会发送一个包, delay 会等待数据，因此操作后它不会发送一个数据包，而是将这段时间内的数据打成一个大的包 keepalive_timeout 0; # 一个keepalive连接在闲置超过一定时间后,服务器关闭这个连接。 client_body_timeout 10; # second HTTP包体 client_header_timeout 10; # second 客户端与服务器建立连接后将开始接收HTTP头部，在这个过程中，如果在一个时间间隔（超时时间）内没有读取到客户端发来的字节，则认为超时，并向客户端返回408($$$"Request timed out")响应。 client_header_buffer_size 1k; # 正常情况下Nginx接收用户请求中HTTP header部分（包括HTTP行和HTTP头部）时分配的内存buffer大小。 large_client_header_buffers 4 4k; # Nginx接收一个超大HTTP头部请求的buffer个数和每个buffer的大小。如果HTTP请求行（如GET/index HTTP/1.1）的大小超过上面的单个buffer，则返回414 output_buffers 2 32k; client_max_body_size 64m; # 浏览器在发送含有较大HTTP包体的请求时，其头部会有一个Content-Length字段，client_max_body_size是用来限制Content-Length所示值的大小的。因此，这个限制包体的配置非常有用处，因为不用等Nginx接收完所有的HTTP包体——这有可能消耗很长时间——就可以告诉用户请求过大不被接受。 client_body_buffer_size 256k; # Nginx接收HTTP包体的内存缓冲区大小。也就是说，HTTP包体会先接收到指定的这块缓存中，之后才决定是否写入磁盘 client_body_in_file_only off; # 当值为非off时，用户请求中的HTTP包体一律存储到磁盘文件中，即使只有0字节也会存储为文件。当请求结束时，如果配置为on，则这个文件不会被删除（该配置一般用于调试、定位问题）​，但如果配置为clean，则会删除该文件。 client_body_in_single_buffer off; # 用户请求中的HTTP包体一律存储到内存buffer中。当然，如果HTTP包体的大小超过了下面client_body_buffer_size设置的值，包体还是会写入到磁盘文件中。 large_client_header_buffers 2k; # HTTP包体会先接收到指定的这块缓存中，之后才决定是否写入磁盘。如果用户请求中含有HTTP头部Content-Length，并且其标识的长度小于定义的buffer大小，那么Nginx会自动降低本次请求所使用的内存buffer，以降低内存消耗。 connection_pool_size 256; # Nginx对于每个建立成功的TCP连接会预先分配一个内存池，上面的size配置项将指定这个内存池的初始大小 request_pool_size 1k; # Nginx开始处理HTTP请求时，将会为每个请求都分配一个内存池，size配置项将指定这个内存池的初始大小 # client_body_temp_path # HTTP包体的临时存放目录 send_timeout 30; # second 发送响应的超时时间，即Nginx服务器向客户端发送了数据包，但客户端一直没有去接收这个数据包。如果某个连接超过send_timeout定义的超时时间，那么Nginx将会关闭这个连接。 reset_timeout_connection on; # 连接超时后将通过向客户端发送RST包来直接重置连接。这个选项打开后，Nginx会在某个连接超时后，不是使用正常情形下的四次握手关闭TCP连接，而是直接向用户发送RST重置包，不再等待用户的应答，直接释放Nginx服务器上关于这个套接字使用的所有缓存（如TCP滑动窗口 keepalive_requests 100; # 一个keepalive连接上默认最多只能发送100个请求。 limit_rate 4k; # 此配置是对客户端请求限制每秒传输的字节数。 limit_rate_after 1m; # 发送的响应长度超过limit_rate_after后才开始限速 ignore_invalid_headers off; # 忽略不合法的头部，默认on，如果将其设置为off，那么当出现不合法的HTTP头部时，Nginx会拒绝服务，并直接向用户发送400(Bad Request)错误。如果将其设置为on，则会忽略此HTTP头部。 underscores_in_headers off; # HTTP头部的名称中不允许带下划线 if_modified_since exact; # 浏览器在第一次请求资源后会缓存响应内容。在后续请求中，浏览器会在请求头中添加 If-Modified-Since，服务器根据该值判断是否需要返回最新的资源内容 # off：表示忽略用户请求中的If-Modified-Since头部。这时，如果获取一个文件，那么会正常地返回文件内容。HTTP响应码通常是200。 # exact：将If-Modified-Since头部包含的时间与将要返回的文件上次修改的时间做精确比较，如果没有匹配上，则返回200和文件的实际内容，如果匹配上，则返回 304 Not Modified # before：是比exact更宽松的比较。只要文件的上次修改时间等于或者早于用户请求中的If-Modified-Since头部的时间，就会向客户端返回304 Not Modified。 log_not_found on; # 此配置项表示当处理用户请求且需要访问文件时，如果没有找到文件，是否将错误日志记录到error.log文件中。 merge_slashes on; # 是否合并相邻的/ resolver 127.0.0.1 127.0.0.2; # DNS 解析地址 resolver_timeout 30; # DNS解析超时时间 server_tokens off; # 表示处理请求出错时是否在响应的Server头部中标明Nginx版本 server { listen 443 default deferred backlog=100 ssl; server_name work.ronie.ren; # 在开始处理一个HTTP请求时，Nginx会取出header头中的Host，与每个server中的server_name进行匹配，以此决定到底由哪一个server块来处理这个请求。有可能一个Host与多个server块中的server_name都匹配，这时就会根据匹配优先级来选择实际处理的server块。server_name与Host的匹配优先级如下：1)首先选择所有字符串完全匹配的server_name，如www.testweb.com。2)其次选择通配符在前面的server_name，如*.testweb.com。3)再次选择通配符在后面的server_name，如www.testweb.*。4)最后选择使用正则表达式才匹配的server_name，如~^\.testweb\.com$。 server_names_hash_bucket_size 1024 # 用于设置存储服务器名称的哈希表的桶大小。 server_names_hash_max_size 512 # 会影响散列表的冲突率。server_names_hash_max_size越大，消耗的内存就越多，但散列key的冲突率则会降低，检索速度也更快。 server_tokens off; keepalive_timeout 10; ssl_certificate /usr/local/lighthouse/softwares/nginx/certificates/work.ronie.ren_bundle.crt; ssl_certificate_key /usr/local/lighthouse/softwares/nginx/certificates/work.ronie.ren.key; ssl_session_timeout 5m; ssl_protocols TLSv1 TLSv1.1 TLSv1.2; ssl_ciphers ECDHE-RSA-AES128-GCM-SHA256:HIGH:!aNULL:!MD5:!RC4:!DHE; ssl_prefer_server_ciphers on; # 反向代理配置 location / { proxy_pass http://workservers; # 设置代理 proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; } access_log "logs/work.ronie.ren.log"; error_log "logs/work.ronie.ren.error.log"; } upstream workservers{ ip_hash; # 据客户端的IP地址计算出一个key，将key按照upstream集群里的上游服务器数量进行取模，然后以取模后的结果把请求转发到相应的上游服务器中。这样就确保了同一个客户端的请求只会转发到指定的上游服务器中。 server 192.168.0.1:8080 weight=5 down; # down表示所在的上游服务器永久下线，只在使用ip_hash配置项时才有用。 server 192.168.0.2:8080 weight=5 max_fails=3 fail_timeout=30s; # 与fail_timeout配合使用，指在fail_timeout时间段内; server 192.168.0.3:8080 weight=5 max_fails=3 fail_timeout=30s; # 如果向当前的上游服务器转发失败次数超过number，则认为在当前的fail_timeout时间段内这台上游服务器不可用。 server 192.168.0.4:8080 weight=5 max_fails=3 fail_timeout=30s backup; # backup：在使用ip_hash配置项时它是无效的。 } }
字段存储，设计必须这么存fieldis_showis_dup_keyxxx00xxx10xxx01sql查询时压缩，c_rule_md5与业务表关联，这是query1SELECT c_rule_md5, field_id, MAX(is_show) AS is_show, MAX(is_dup_key) AS is_dup_key FROM t_rel_complicated_rules_fields trcrf WHERE trcrf.is_dup_key != 1字段信息和字段表和表基本信息表连接查询，获取字段所属表名称，字段名称，字段中文名等，这是query5SELECT ttf.column_name_cn, ttf.column_name, ttbi.table_name, ttbi.table_schema, ttbi.table_schema_cn, ttbi.table_name_cn, query1.c_rule_md5, query1.is_show, query1.is_dup_key, ttf.id AS field_id FROM ( SELECT c_rule_md5, field_id, MAX( is_show ) AS is_show, MAX( is_dup_key ) AS is_dup_key FROM t_rel_complicated_rules_fields trcrf WHERE trcrf.is_dup_key != 1 GROUP BY c_rule_md5, field_id ) query1 LEFT JOIN t_table_field ttf ON query1.field_id = ttf.id LEFT JOIN t_table_base_info ttbi ON ttbi.table_schema = ttf.table_schema AND ttbi.table_name = ttf.table_name WHERE ttbi.table_schema = 'db_enterprise' AND ttbi.table_name = 't_address' AND query1.field_id = 396 )查最新一条报告，这是query3SELECT tcrr.id, tcrr.rule_md5, tcrr.is_alarming, tcrr.create_time AS last_execute_time FROM t_complicated_rule_report tcrr JOIN ( SELECT rule_md5, MAX( create_time ) AS max_create_time FROM t_complicated_rule_report GROUP BY rule_md5 ) query2 ON tcrr.rule_md5 = query2.rule_md5 AND tcrr.create_time = query2.max_create_time最新一条报告和主表连接，tcrc与query3左连接，生成query4SELECT tcrc.id AS id, tcrc.rule_name, tcrc.rule_md5, tcrc.rule_type_id AS rule_type_id, tcrc.create_user, tcrc.create_time, tcrc.cron_type, tcrc.rule_status AS is_online, tcrc.rule_update_time, tcrc.online_approve_id, query3.is_alarming AS is_alarming, query3.last_execute_time, tcrc.try_running_percent FROM t_complicated_rule_config tcrc LEFT JOIN ( SELECT tcrr.id, tcrr.rule_md5, tcrr.is_alarming, tcrr.create_time AS last_execute_time FROM t_complicated_rule_report tcrr JOIN ( SELECT rule_md5, MAX( create_time ) AS max_create_time FROM t_complicated_rule_report GROUP BY rule_md5 ) query2 ON tcrr.rule_md5 = query2.rule_md5 AND tcrr.create_time = query2.max_create_time ) query3 ON query3.rule_md5 = tcrc.rule_md5 ) query4query4与query5进行内连接，生成query6( SELECT query4.*, query5.column_name_cn, query5.column_name, query5.table_name, query5.table_schema, query5.table_schema_cn, query5.table_name_cn, query5.is_show, query5.is_dup_key, query5.field_id FROM (...) query4 INNER JOIN (...) query5 ON query4.rule_md5 = query5.c_rule_md5 ) query6mysql 5.7没有row_number()，需要通过mysql变量生成row_number( SELECT @row_number := 0, @current_md5 := '', @is_show_sum := 0 ) AS vars使用变量 @row_number、@current_md5 和 @is_show_sum 来实现分组和过滤逻辑，生成query7，在这里的目的是生成每个分组内的row_number并判断每个分组内的is_show是否全部为0；注意query6.is_show DESC，可以确保分组内存在is_show=1的值在前，则存在is_show=1分组内每行的is_show_sum>0；SELECT query6.*, @row_number :=IF( @current_md5 = query6.rule_md5, @row_number + 1, 1 ) AS row_number, @is_show_sum := IF( @current_md5 = query6.rule_md5, @is_show_sum + query6.is_show, query6.is_show ) AS is_show_sum, @current_md5 := query6.rule_md5 FROM (...) query6, ( SELECT @row_number := 0, @current_md5 := '', @is_show_sum := 0 ) AS vars ORDER BY query6.rule_md5 ASC, query6.is_show DESC ) query7如果分组内 is_show 有一个不为0，则保留整个分组。如果分组内 is_show 全为0，则只保留分组中的第一行（即 row_number = 1）。SELECT * FROM query7 WHERE ( query7.is_show_sum > 0 OR ( query7.is_show_sum = 0 AND query7.row_number = 1 ))总体：SELECT query7.* from ( select query6.*, @row_number := IF(@current_md5 = query6.rule_md5, @row_number + 1, 1) AS row_number, @is_show_sum := IF(@current_md5 = query6.rule_md5, @is_show_sum + query6.is_show, query6.is_show) AS is_show_sum, @current_md5 := query6.rule_md5 from ( SELECT query4.*, query5.column_name_cn, query5.column_name, query5.table_name, query5.table_schema, query5.table_schema_cn, query5.table_name_cn, query5.is_show, query5.is_dup_key, query5.field_id FROM ( SELECT tcrc.id as id, tcrc.rule_name, tcrc.rule_md5, tcrc.rule_type_id as rule_type_id, tcrc.create_user, tcrc.create_time, tcrc.cron_type, tcrc.rule_status as is_online, tcrc.rule_update_time, tcrc.online_approve_id, query3.is_alarming as is_alarming, query3.last_execute_time, tcrc.try_running_percent FROM t_complicated_rule_config tcrc LEFT JOIN ( SELECT tcrr.id, tcrr.rule_md5, tcrr.is_alarming, tcrr.create_time as last_execute_time FROM t_complicated_rule_report tcrr JOIN ( SELECT rule_md5, MAX(create_time) AS max_create_time FROM t_complicated_rule_report GROUP BY rule_md5 ) query2 ON tcrr.rule_md5 = query2.rule_md5 AND tcrr.create_time = query2.max_create_time ) query3 ON query3.rule_md5 = tcrc.rule_md5 ) query4 INNER JOIN ( SELECT ttf.column_name_cn, ttf.column_name, ttbi.table_name, ttbi.table_schema, ttbi.table_schema_cn, ttbi.table_name_cn, query1.c_rule_md5, query1.is_show, query1.is_dup_key, ttf.id as field_id FROM ( SELECT c_rule_md5, field_id, MAX(is_show) AS is_show, MAX(is_dup_key) AS is_dup_key FROM t_rel_complicated_rules_fields trcrf WHERE trcrf.is_dup_key != 1 GROUP BY c_rule_md5, field_id ) query1 LEFT JOIN t_table_field ttf ON query1.field_id = ttf.id LEFT JOIN t_table_base_info ttbi ON ttbi.table_schema = ttf.table_schema AND ttbi.table_name = ttf.table_name WHERE ttbi.table_schema = 'db_enterprise' AND ttbi.table_name = 't_address' AND query1.field_id = 396 ) query5 ON query4.rule_md5 = query5.c_rule_md5 ) query6, ( SELECT @row_number := 0, @current_md5 := '', @is_show_sum := 0) AS vars order by query6.rule_md5 asc, query6.is_show desc ) query7 where (query7.is_show_sum > 0 OR (query7.is_show_sum = 0 AND query7.row_number = 1)) AND rule_type_id = 4 AND query7.is_online = 1 AND (query7.is_alarming = 0 or query7.is_alarming is NULL) AND query7.try_running_percent = 100
docker kafka 配置文件version: '3.8' name: "kafka-service" # https://docs.docker.com/compose/how-tos/project-name/ services: kafka-server0: image: bitnami/kafka:latest hostname: kafka0 container_name: kafka0 ports: - "9092:9094" environment: KAFKA_CFG_NODE_ID: 0 KAFKA_CFG_BROKER_ID: 0 KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://0.0.0.0:9094 KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://kafka0:9092,EXTERNAL://192.168.60.134:9092 KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 1@kafka1:9093,2@kafka2:9093,0@kafka0:9093 KAFKA_CFG_PROCESS_ROLES: broker,controller # KAFKA_CFG_LOG_DIRS: /var/lib/kafka/data KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR: 3 KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3 KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR: 2 KAFKA_CFG_MIN_INSYNC_REPLICAS: 1 KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: true KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER KAFKA_KRAFT_CLUSTER_ID: d6c6ea9d1bfe423ebdc74579b10c61f0 networks: kafka: ipv4_address: 172.16.0.2 volumes: - ~/kafka-data/0:/bitnami/kafka kafka-server1: image: bitnami/kafka:latest hostname: kafka1 container_name: kafka1 ports: - "9094:9094" environment: KAFKA_CFG_NODE_ID: 1 KAFKA_CFG_BROKER_ID: 1 KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://0.0.0.0:9094 KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:9092,EXTERNAL://192.168.60.134:9094 KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 1@kafka1:9093,2@kafka2:9093,0@kafka0:9093 KAFKA_CFG_PROCESS_ROLES: broker,controller # KAFKA_CFG_LOG_DIRS: /var/lib/kafka/data KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR: 3 KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3 KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR: 2 KAFKA_CFG_MIN_INSYNC_REPLICAS: 1 KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: true KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER KAFKA_KRAFT_CLUSTER_ID: d6c6ea9d1bfe423ebdc74579b10c61f0 networks: kafka: ipv4_address: 172.16.0.3 volumes: - ~/kafka-data/1:/bitnami/kafka kafka-server2: image: bitnami/kafka:latest hostname: kafka2 container_name: kafka2 ports: - "9096:9094" environment: KAFKA_CFG_NODE_ID: 2 KAFKA_CFG_BROKER_ID: 2 KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://0.0.0.0:9094 KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://kafka2:9092,EXTERNAL://192.168.60.134:9096 KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 1@kafka1:9093,2@kafka2:9093,0@kafka0:9093 KAFKA_CFG_PROCESS_ROLES: broker,controller # KAFKA_CFG_LOG_DIRS: /var/lib/kafka/data KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR: 3 KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3 KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR: 2 KAFKA_CFG_MIN_INSYNC_REPLICAS: 1 KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: true KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER KAFKA_KRAFT_CLUSTER_ID: d6c6ea9d1bfe423ebdc74579b10c61f0 networks: kafka: ipv4_address: 172.16.0.4 volumes: - ~/kafka-data/2:/bitnami/kafka networks: kafka: driver: bridge name: kafka-net ipam: config: - subnet: 172.16.0.0/24 gateway: 172.16.0.1 KAFKA_CFG_LISTENERS：配置的监听器将被用于监听网络请求，listener_name://host_name:port,listener_name2://host_name2:port2。可以同时配置多个，并且用逗号隔开，监听器的名称和端口必须是唯一的，host_name如果为空，则会绑定到默认的接口(网卡)。KAFKA_CFG_ADVERTISED_LISTENERS：ADVERTISED_LISTENERS 配置 Kafka broker 向客户端（生产者和消费者）报告的地址。这是客户端用来连接到 broker 的地址。例如：ADVERTISED_LISTENERS=PLAINTEXT://broker-hostname:9092 表示 broker 将向客户端报告 broker-hostname:9092 作为连接地址。KAFKA_CFG_PROCESS_ROLES：如果PROCESS_ROLES = Broker，服务器在KRaft模式中充当 Broker。如果PROCESS_ROLES = Controller，服务器在KRaft模式下充当 Controller。如果PROCESS_ROLES = Broker,Controller，服务器在KRaft模式中同时充当 Broker 和Controller。如果PROCESS_ROLES没有设置，那么集群就假定是运行在ZooKeeper模式下。KAFKA_CFG_CONTROLLER_QUORUM_VOTERS：标识有哪些节点是 Quorum 的投票者节点。所有想成为控制器的节点都需要包含在这个配置里面。需要包含每个节点的id。如KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 1@kafka1:9093,2@kafka2:9093,0@kafka0:9093KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR：用于指定内部主题 __consumer_offsets 的副本因子。__consumer_offsets 主题用于存储消费者组的偏移量信息。设置 KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR: 3 表示 __consumer_offsets 主题将有 3 个副本。这意味着即使有两个 broker 发生故障，仍然可以从剩余的一个副本中读取消费者偏移量信息。KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR：指定内部主题 __transaction_state 的副本因子。__transaction_state 主题用于存储事务状态信息。KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR：指定了事务状态日志主题的最小同步副本数（ISR）。这个参数确保在提交事务时，至少有指定数量的副本是同步的。说一说你对 Kafka 中 ISR 的理解-腾讯云开发者社区-腾讯云KAFKA_CFG_MIN_INSYNC_REPLICAS：指定在写操作（例如生产消息）被认为成功之前，必须确认写入的最小同步副本（ISR）数量。问题：在外部网络能够正常使用KafkaProducer向kafka docker集群生产消息，但KafkaConsumer不能正常使用group_id消费消息（而不指定group_id，却能消费消息）。问题无法复现，在解决NotLeaderForPartitionError问题后，不在出现。挂载卷问题（待解决）：kafka0 | mkdir: cannot create directory '/bitnami/kafka/config': Permission denied kafka2 | mkdir: cannot create directory '/bitnami/kafka/config': Permission denied kafka1 | mkdir: cannot create directory '/bitnami/kafka/config': Permission deniedNOTE: As this is a non-root container, the mounted files and directories must have the proper permissions for the UID 1001.sudo chmod 776 -R kafka-data/：为其他用户组给出可读可写权限。Use non-root containersWhy non-root containers are important for securityWork With Non-Root ContainersRunning non-root containers on Openshift生产者 NotLeaderForPartitionError：KAFKA_CFG_ADVERTISED_LISTENERS配置错误：错误配置：kafka-server1: image: bitnami/kafka:latest hostname: kafka1 container_name: kafka1 ports: - "9092:9094" environment: KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:9094 KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:9092,EXTERNAL://localhost:9094正确配置：kafka-server1: image: bitnami/kafka:latest hostname: kafka1 container_name: kafka1 ports: - "9092:9094" environment: KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://0.0.0.0:9094 KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:9092,EXTERNAL://192.168.60.134:9092原因：Note: To connect from an external machine, change localhost above to your host's external IP/hostname and include EXTERNAL://0.0.0.0:9094 in KAFKA_CFG_LISTENERS to allow for remote connections.KAFKA_CFG_ADVERTISED_LISTENERS:PLAINTEXT://kafka1:9092,EXTERNAL://localhost:9094 修改为PLAINTEXT://kafka1:9092,EXTERNAL://192.168.60.134:9092192.168.60.134为docker宿主机IP，9092为docker宿主机映射端口(.venv) ronie@ronie:~/PycharmProjects/evilbox$ docker ps CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES f2509ff75469 bitnami/kafka:latest "/opt/bitnami/script…" 6 seconds ago Up 6 seconds 9092/tcp, 0.0.0.0:9092->9094/tcp, [::]:9092->9094/tcp kafka0 因为docker的网络拓扑是这样的graph TD subgraph Docker_Host direction TB br0[Bridge Network: kafka-net] kafka0[Kafka Broker 0] kafka1[Kafka Broker 1] kafka2[Kafka Broker 2] br0 --- kafka0 br0 --- kafka1 br0 --- kafka2 end subgraph External_Network direction TB ext[External Client ] end ext --- br0
RedisRedis-Server通过bitnami/redis:7.4.1实现 redis 主从集群version: '3.8' name: 'redis-service' services: redis-master: image: 'bitnami/redis:7.4.1' container_name: redis-master hostname: redis-master expose: - '6379' environment: - REDIS_REPLICATION_MODE=master - REDIS_PASSWORD=OeLFo2b3GVXKj0Ren1Dh - REDIS_MASTER_PORT_NUMBER=6379 networks: redis: ipv4_address: 192.168.0.2 redis-replica: image: 'bitnami/redis:7.4.1' deploy: replicas: 2 depends_on: - redis-master environment: - REDIS_REPLICATION_MODE=slave - REDIS_MASTER_HOST=192.168.0.2 - REDIS_MASTER_PORT_NUMBER=6379 - REDIS_MASTER_PASSWORD=OeLFo2b3GVXKj0Ren1Dh - REDIS_PASSWORD=OeLFo2b3GVXKj0Ren1Dh networks: redis: networks: redis: driver: bridge name: redis-net ipam: config: - subnet: 192.168.0.0/24 gateway: 192.168.0.1 Redis-Sentinel为保证redis集群高可用，添加redis sentinel监控master节点。version: '3.8' name: 'redis-sentinel-cluster' services: redis-sentinel: image: 'bitnami/redis-sentinel:6.2.16' deploy: replicas: 3 environment: REDIS_SENTINEL_PORT_NUMBER: '26379' REDIS_MASTER_HOST: '192.168.0.2' REDIS_MASTER_PORT_NUMBER: '6379' REDIS_SENTINEL_QUORUM: '2' # 在进行故障转移之前，必须同意主节点失效的 Sentinel 实例的最小数量。 REDIS_MASTER_PASSWORD: 'OeLFo2b3GVXKj0Ren1Dh' REDIS_SENTINEL_DOWN_AFTER_MILLISECONDS: '5000' REDIS_SENTINEL_FAILOVER_TIMEOUT: '5000' REDIS_MASTER_SET: 'mymaster' networks: - redis networks: redis: name: 'redis-net' external: true测试后发现，redis sentinel 能在master节点失效后选举新master节点，但是在docker-compose中只暴露了master节点的6379端口。于是在redis-net网络中添加nginx反代。nginx反代（不可用）nginx-reverse-proxy: image: 'bitnami/nginx:1.27' volumes: - ./nginx.conf:/opt/bitnami/nginx/conf/stream_server_blocks/my_server_block.conf environment: NGINX_ENABLE_STREAM: yes ports: - '6379:6379' networks: redis: depends_on: - redis-replica添加反代后，发现slave节点不能向master节点转发命令（记错了，这个是redis分片集群的功能，分片节点计算键的插槽，该插槽不归该节点管就转发给该管这个插槽的节点）。放弃反代，直接将测试代码接入redis-net网络。问题发现容器内无法访问外部网络(.venv) ronie@ronie:~/PycharmProjects/evilbox$ docker attach 81 / # ping www.baidu.com PING www.baidu.com (180.101.50.188): 56 data bytes ^C --- www.baidu.com ping statistics --- 2 packets transmitted, 0 packets received, 100% packet loss发现docker0网卡down且未被分配IP。(.venv) ronie@ronie:~/PycharmProjects/evilbox$ ip addr show docker0 3: docker0: <NO-CARRIER,BROADCAST,MULTICAST,UP> mtu 1500 qdisc noqueue state DOWN group default link/ether 02:42:e2:f4:ca:4c brd ff:ff:ff:ff:ff:ff 猜测可能是root mode docker 和 rootless mode docker 冲突。停止root mode docker。(.venv) ronie@ronie:~/PycharmProjects/evilbox$ ip addr show docker0 3: docker0: <NO-CARRIER,BROADCAST,MULTICAST,UP> mtu 1500 qdisc noqueue state DOWN group default link/ether 02:42:f3:75:a1:fd brd ff:ff:ff:ff:ff:ff inet 172.17.0.1/16 brd 172.17.255.255 scope global docker0 valid_lft forever preferred_lft forever 分配了IP，但仍然是down（应该是down，docker0在root mode下使用）。检查容器网络设置 docker inspect 89，发现容器被设置了代理？？？不知道什么时候修改的，该回来。Proxy configuration | Docker Docs"Env": [ "NO_PROXY=localhost, 127.0.0.1, 10.96.0.0/12, 192.168.59.0/24, 192.168.49.0/24, 192.168.39.0/24", "no_proxy=localhost, 127.0.0.1, 10.96.0.0/12, 192.168.59.0/24, 192.168.49.0/24, 192.168.39.0/24", "HTTP_PROXY=http://127.0.0.1:2081", "http_proxy=http://127.0.0.1:2081", "HTTPS_PROXY=http://127.0.0.1:2081", "https_proxy=http://127.0.0.1:2081" ],注意到4404 /proc/self/exe --state-dir=/run/user/1000/dockerd-rootless --net=slirp4netns ，发现docker在rootless模式下是不用docker0这个网卡的。rootlesskit/docs/network.md at v2.0.0 · rootless-containers/rootlesskitRootless mode | Docker DocsPing does not work when /proc/sys/net/ipv4/ping_group_range is set to 1 0:(.venv) ronie@ronie:/proc/sys/net/ipv4$ pwd /proc/sys/net/ipv4 (.venv) ronie@ronie:/proc/sys/net/ipv4$ cat ping_group_range 1 0Rootless mode | Docker DocsOn some distributions, ping does not work by default.Add net.ipv4.ping_group_range = 0 2147483647 to /etc/sysctl.conf (or /etc/sysctl.d) and run sudo sysctl --system to allow using ping.rootless docker 网络 | 今天我学了什么总的来说是代理问题。clientimport typing from redis.sentinel import Sentinel from redis import Redis import redis import string import random class RedisClient(Redis): def __init__( self, sentinels: typing.Union[typing.List[typing.Tuple[str, int]], Sentinel], **kwargs ): if isinstance(sentinels, Sentinel): self.sentinels = sentinels else: self.sentinels = Sentinel(sentinels, socket_timeout=0.1) self.client_kwargs = kwargs host, port = self.sentinels.discover_master("mymaster") super().__init__(host=host, port=port, **kwargs) def renew(self): host, port = self.sentinels.discover_master("mymaster") super().__init__(host=host, port=port, **self.client_kwargs) return self def __getattribute__(self, item): attr = super().__getattribute__(item) if item != 'ensure_client' and callable(attr): attr = self.ensure_client(attr) return attr @staticmethod def ensure_client(func): def wrap(*args, **kwargs): self = args[0] if args else None if self is None: return func(*args, **kwargs) for i in range(3): try: ret = func(*args, **kwargs) break except redis.exceptions.ConnectionError: self.renew() else: raise redis.exceptions.ConnectionError return ret return wrap if __name__ == '__main__': r = RedisClient( sentinels=[ ('redis-sentinel-cluster-sentinel-1', 26379), ('redis-sentinel-cluster-sentinel-2', 26379), ('redis-sentinel-cluster-sentinel-3', 26379), ], password='OeLFo2b3GVXKj0Ren1Dh' ) r.lpush("test", "".join(random.choices(string.printable, k=20))) print(r.rpop("test"))
2645. 构造有效字符串的最少插入数2645. 构造有效字符串的最少插入数class Solution: def addMinimum(self, word: str) -> int: res = ord(word[0]) - ord(word[-1]) + 2 for i in range(1, len(word)): res += (ord(word[i]) - ord(word[i - 1]) + 2) % 3 return res1020. 飞地的数量1020. 飞地的数量from typing import List class UnionFindSet: def __init__(self): self.father = {} def find(self, val): if val not in self.father: self.father[val] = val if self.father[val] != val: self.father[val] = self.find(self.father[val]) return self.father[val] def union(self, val1, val2): root_val1 = self.find(val1) root_val2 = self.find(val2) self.father[root_val2] = root_val1 class Solution: def __init__(self): self.direction = [[-1, 0], [1, 0], [0, -1], [0, 1]] def numEnclaves(self, grid: List[List[int]]) -> int: uf = UnionFindSet() for i in (0, len(grid) - 1): for j in range(len(grid[i])): self.set_zero(grid, i, j) for i in range(len(grid)): for j in (0, len(grid[i]) - 1): self.set_zero(grid, i, j) for i in range(1, len(grid) - 1): for j in range(1, len(grid[i]) - 1): if grid[i][j] == 1: uf.find((i, j)) grid[i][j] = 2 for x, y in self.direction: if grid[i + x][j + y] == 1: uf.union((i, j), (i + x, j + y)) return len(uf.father) def set_zero(self, grid, x, y): if grid[x][y] == 0: return grid[x][y] = 0 for i, j in self.direction: if x + i < 0 or x + i >= len(grid) or y + j < 0 or y + j >= len(grid[x + i]): continue self.set_zero(grid, x + i, y + j) 1559. 二维网格图中探测环1559. 二维网格图中探测环from typing import List class UnionFindSet: def __init__(self): self.father = {} def find(self, val): if val not in self.father: self.father[val] = val if self.father[val] != val: self.father[val] = self.find(self.father[val]) return self.father[val] def union(self, val1, val2): root_val1 = self.find(val1) root_val2 = self.find(val2) if root_val1 == root_val2: return True self.father[root_val1] = self.find(root_val2) return False class Solution: def __init__(self): self.direction = [[-1, 0], [1, 0], [0, -1], [0, 1]] def containsCycle(self, grid: List[List[str]]) -> bool: uf = UnionFindSet() for i in range(len(grid)): for j in range(len(grid[i])): uf.find((i, j)) for x, y in self.direction: if i + x < 0 or i + x >= len(grid) or j + y < 0 or j + y >= len(grid[i + x]): continue if grid[i][j] == grid[i + x][j + y]: if uf.union((i, j), (i + x, j + y)): return True grid[i][j] = '' return False2241. 设计一个 ATM 机器2241. 设计一个 ATM 机器from typing import List class ATM: def __init__(self): self.moneys = [0, 0, 0, 0, 0] self.pos = {20: 0, 50: 1, 100: 2, 200: 3, 500: 4} self.order = {0: 20, 1: 50, 2: 100, 3: 200, 4: 500} def deposit(self, banknotesCount: List[int]) -> None: for i, count in enumerate(banknotesCount): self.moneys[i] += count def withdraw(self, amount: int) -> List[int]: ans = [0, 0, 0, 0, 0] for i in range(len(self.moneys) - 1, -1, -1): ans[i] += amount // self.order[i] if (amount // self.order[i]) <= self.moneys[i] else self.moneys[i] self.moneys[i] -= ans[i] amount -= ans[i] * self.order[i] if amount: self.deposit(ans) return [-1] return ans1381. 设计一个支持增量操作的栈1381. 设计一个支持增量操作的栈class CustomStack: def __init__(self, maxSize: int): self.stack = [] self.max_size = maxSize self.top = 0 def push(self, x: int) -> None: if len(self.stack) == self.max_size: return self.stack.append(x - self.top) self.top += self.stack[-1] def pop(self) -> int: if not self.stack: return -1 ans = self.top self.top -= self.stack.pop() return ans def increment(self, k: int, val: int) -> None: if not self.stack: return self.stack[0] += val if k < len(self.stack): self.stack[k] -= val else: self.top += val1305. 两棵二叉搜索树中的所有元素1305. 两棵二叉搜索树中的所有元素生成器class Solution: def getAllElements(self, root1: TreeNode, root2: TreeNode) -> List[int]: tree1 = self.dfs(root1) tree2 = self.dfs(root2) tree1_val = next(tree1, None) tree2_val = next(tree2, None) ans = [] while tree1_val is not None or tree2_val is not None: if tree1_val is None: ans.append(tree2_val) tree2_val = next(tree2, None) elif tree2_val is None: ans.append(tree1_val) tree1_val = next(tree1, None) elif tree1_val < tree2_val: ans.append(tree1_val) tree1_val = next(tree1, None) else: ans.append(tree2_val) tree2_val = next(tree2, None) return ans def dfs(self, node: TreeNode): if node is None: return if node.left: yield from self.dfs(node.left) yield node.val if node.right: yield from self.dfs(node.right)
735. 小行星碰撞735. 小行星碰撞class Solution: def asteroidCollision(self, asteroids: List[int]) -> List[int]: stack = [asteroids[0]] for i in range(1, len(asteroids)): if stack and stack[-1] < 0: stack.append(asteroids[i]) elif stack and asteroids[i] < 0: while stack and -asteroids[i] > stack[-1] > 0: stack.pop() if stack and -asteroids[i] == stack[-1]: stack.pop() continue if stack and -asteroids[i] < stack[-1]: continue stack.append(asteroids[i]) else: stack.append(asteroids[i]) return stack895. 最大频率栈895. 最大频率栈优先队列，时间复杂度$O(nlogn)$；题解将栈序列分解成为多个频率不同的栈序列，时间复杂度$O(1)$。from collections import defaultdict from typing import Dict import heapq class FreqStack: def __init__(self): self.frequency: Dict[int, int] = defaultdict(int) self.heap = [] self.index = 0 def push(self, val: int) -> None: self.frequency[val] += 1 self.index -= 1 heapq.heappush(self.heap, (-self.frequency[val], self.index, val)) def pop(self) -> int: freq, index, val = heapq.heappop(self.heap) self.frequency[val] -= 1 return val1319. 连通网络的操作次数1319. 连通网络的操作次数深度优先搜索，还可使用并查集from typing import List, Dict from collections import defaultdict class Solution: def makeConnected(self, n: int, connections: List[List[int]]) -> int: if len(connections) < n - 1: return -1 conn: Dict[int, list] = defaultdict(list) for src, dst in connections: conn[src].append(dst) conn[dst].append(src) checked = set() ans = 0 for i in range(n): if i not in checked: self.dfs(conn, i, checked) ans += 1 return ans - 1 def dfs(self, conn, start, checked): checked.add(start) for computer in conn[start]: if computer in checked: continue self.dfs(conn, computer, checked)LCR 066. 键值映射LCR 066. 键值映射字典树class Node: __slots__ = ["child", 'char', 'val'] def __init__(self, c, val=0): self.child = {} self.char = c self.val = val class MapSum: def __init__(self): """ Initialize your data structure here. """ self.tree = Node(None) def insert(self, key: str, val: int) -> None: node = self.tree for c in key: if c in node.child: node = node.child[c] else: new_node = Node(c) node.child[c] = new_node node = new_node node.val = val def sum(self, prefix: str) -> int: node = self.tree for c in prefix: if c not in node.child: return 0 node = node.child[c] return self.dfs(node) def dfs(self, node): ret = node.val for char, child_node in node.child.items(): ret += self.dfs(child_node) return ret 684. 冗余连接684. 冗余连接class UnionFindSet: def __init__(self): self.parent = {} def find(self, x): if x not in self.parent: self.parent[x] = x elif self.parent[x] != x: self.parent[x] = self.find(self.parent[x]) return self.parent[x] def union(self, x, y): root_x = self.find(x) root_y = self.find(y) if root_x == root_y: return False self.parent[root_x] = root_y return True class Solution: def findRedundantConnection(self, edges: List[List[int]]) -> List[int]: uf = UnionFindSet() for x, y in edges: if not uf.union(x, y): return [x, y]
2696. 删除子串后的字符串最小长度2696. 删除子串后的字符串最小长度class Solution: def minLength(self, s: str) -> int: while True: tmp_s = s s = s.replace('AB', '').replace('CD', '') if tmp_s == s: break return len(s)787. K 站中转内最便宜的航班787. K 站中转内最便宜的航班from typing import List, Dict from collections import defaultdict import heapq class Solution: def findCheapestPrice(self, n: int, flights: List[List[int]], src: int, dst: int, k: int) -> int: edges = flights flights: Dict[int, list] = defaultdict(list) for s, d, c in edges: flights[s].append((c, d)) priority_queue = [(0, src, 0, [])] arrived_airport = set() while priority_queue: cost, airport, transfer_cnt, path = heapq.heappop(priority_queue) arrived_airport.add(airport) if airport == dst: return cost if transfer_cnt == k + 1: for item in path: arrived_airport.discard(item) continue for c, d in flights[airport]: if d in arrived_airport: continue path.append(d) heapq.heappush(priority_queue, (cost + c, d, transfer_cnt + 1, path[:])) path.pop() return -1 LCP 67. 装饰树LCP 67. 装饰树class Solution: def expandBinaryTree(self, root: Optional[TreeNode]) -> Optional[TreeNode]: self.dfs(root) return root def dfs(self, node: Optional[TreeNode]): if node.left: insert_node = TreeNode(-1) insert_node.left = node.left node.left = insert_node self.dfs(insert_node.left) if node.right: insert_node = TreeNode(-1) insert_node.right = node.right node.right = insert_node self.dfs(insert_node.right)236. 二叉树的最近公共祖先236. 二叉树的最近公共祖先class Solution: def lowestCommonAncestor(self, root: 'TreeNode', p: 'TreeNode', q: 'TreeNode') -> 'TreeNode': p_path = [] self.dfs(root, p, p_path) q_path = [] self.dfs(root, q, q_path) index = 0 while index < len(p_path) and index < len(q_path): if p_path[index].val != q_path[index].val: return p_path[index - 1] index += 1 return p_path[-1] if len(p_path) < len(q_path) else q_path[-1] def dfs(self, node, search_node, path: list): if node.val == search_node.val: path.append(node) return True path.append(node) if node.left and self.dfs(node.left, search_node, path): return True if node.right and self.dfs(node.right, search_node, path): return True path.pop() return False 1472. 设计浏览器历史记录1472. 设计浏览器历史记录对顶栈from collections import deque class BrowserHistory: def __init__(self, homepage: str): self.stack_before = deque([homepage]) self.stack_after = deque() def visit(self, url: str) -> None: self.stack_before.append(url) self.stack_after.clear() def back(self, steps: int) -> str: for i in range(steps): if len(self.stack_before) == 1: break self.stack_after.appendleft(self.stack_before.pop()) return self.stack_before[-1] def forward(self, steps: int) -> str: for i in range(steps): if not self.stack_after: break self.stack_before.append(self.stack_after.popleft()) return self.stack_before[-1]面试题 10.10. 数字流的秩面试题 10.10. 数字流的秩import bisect class StreamRank: def __init__(self): self.stream = [] def track(self, x: int) -> None: bisect.insort(self.stream, x) def getRankOfNumber(self, x: int) -> int: return bisect.bisect_right(self.stream, x)
2707. 字符串中的额外字符2707. 字符串中的额外字符class Solution: def minExtraChar(self, s: str, dictionary: List[str]) -> int: dictionary = set(dictionary) dp = [0 for i in range(len(s) + 1)] for i in range(1, len(s) + 1): dp[i] = dp[i - 1] + 1 for j in range(i - 1, -1, -1): if s[j:i] in dictionary: dp[i] = min(dp[i], dp[j]) return dp[-1]841. 钥匙和房间841. 钥匙和房间class Solution: def canVisitAllRooms(self, rooms: List[List[int]]) -> bool: need_enter = {i for i in range(len(rooms))} have_keys = {0} while have_keys: key = have_keys.pop() if key in need_enter: need_enter.discard(key) have_keys.update(set(rooms[key])) return not bool(need_enter)1261. 在受污染的二叉树中查找元素1261. 在受污染的二叉树中查找元素from typing import List, Optional from tools import treeGenerate import json # Definition for a binary tree node. class TreeNode: def __init__(self, val=0, left=None, right=None): self.val = val self.left = left self.right = right from collections import deque class FindElements: def __init__(self, root: Optional[TreeNode]): root.val = 0 self.root = self.recover(root) def recover(self, node: Optional[TreeNode]): if node.left: node.left.val = 2 * node.val + 1 self.recover(node.left) if node.right: node.right.val = 2 * node.val + 2 self.recover(node.right) return node def find(self, target: int) -> bool: path = deque() while target != 0: path.appendleft(target) target = (target - 1) // 2 node = self.root while path: val = path.popleft() if node.val * 2 + 1 == val and node.left: node = node.left elif node.val * 2 + 2 == val and node.right: node = node.right else: return False return True 1905. 统计子岛屿1905. 统计子岛屿class Solution: def countSubIslands(self, grid1: List[List[int]], grid2: List[List[int]]) -> int: m = len(grid1) n = len(grid1[0]) not_check_island = set() for i in range(m): for j in range(n): if grid1[i][j] == 0 and grid2[i][j] == 1: not_check_island.add((i, j)) island = [] for i in range(len(grid2)): for j in range(len(grid2[i])): if grid2[i][j] == 1: island_area = set() ret = self.dfs(grid2, i, j, island_area, not_check_island) if ret: island.append(island_area) return len(island) def dfs(self, grid, i, j, island_area: set, not_check_island: set): if (i, j) in not_check_island: return False if grid[i][j] != 1: return True island_area.add((i, j)) grid[i][j] = 2 ret = [] for x, y in [[-1, 0], [1, 0], [0, -1], [0, 1]]: if i + x < 0 or i + x >= len(grid) or j + y < 0 or j + y >= len(grid[x]): continue ret.append(self.dfs(grid, i + x, j + y, island_area, not_check_island)) return all(ret)1268. 搜索推荐系统1268. 搜索推荐系统字典树 + 二分法import bisect class Node: __slots__ = ['char', 'child', 'end_mark', 'child_vals'] def __init__(self, char): self.char = char self.child_vals = [] self.child = {} self.end_mark = False class DictionaryTree: def __init__(self): self.root = Node(None) def add_word(self, word): node = self.root for c in word: if c in node.child: node = node.child.get(c) else: new = Node(c) node.child[c] = new bisect.insort(node.child_vals, c) node = new node.end_mark = True def dfs(self, node: Optional[Node], words: list, chars: list, limit: int): if node is None: return if len(words) >= limit: return if not node.child: words.append(''.join(chars)) return if node.end_mark: words.append(''.join(chars)) for k in node.child_vals: chars.append(k) self.dfs(node.child[k], words, chars, limit) chars.pop() class Solution: def suggestedProducts(self, products: List[str], searchWord: str) -> List[List[str]]: dt = DictionaryTree() for word in products: dt.add_word(word) ans = [] node = dt.root for i, c in enumerate(searchWord, start=1): if node: node = node.child.get(c) candidate_words = [] dt.dfs(node, candidate_words, [searchWord[:i]], 3) ans.append(candidate_words) return ans743. 网络延迟时间743. 网络延迟时间优先队列或使用最短路算法import heapq from collections import defaultdict class Solution: def networkDelayTime(self, times: List[List[int]], n: int, k: int) -> int: net: Dict[int, list] = defaultdict(list) nodes = {i + 1 for i in range(n)} for u, v, w in times: net[u].append((v, w)) priority_queue = [(0, k)] t = 0 while priority_queue and nodes: t, node = heapq.heappop(priority_queue) nodes.discard(node) for v, w in net[node]: if v in nodes: heapq.heappush(priority_queue, (t + w, v)) if nodes: return -1 return t
764. 最大加号标志764. 最大加号标志class Solution: def orderOfLargestPlusSign(self, n: int, mines: List[List[int]]) -> int: dp = [[[0, 0, 0, 0] for i in range(n)] for j in range(n)] mines = set(list(map(tuple, mines))) for i in range(n): for j in range(n): if (i, j) in mines: dp[i][j][0] = 0 dp[i][j][1] = 0 else: dp[i][j][0] = dp[i - 1][j][0] + 1 if i - 1 >= 0 else 1 dp[i][j][1] = dp[i][j - 1][1] + 1 if j - 1 >= 0 else 1 for i in range(n - 1, -1, -1): for j in range(n - 1, -1, -1): if (i, j) in mines: dp[i][j][2] = 0 dp[i][j][3] = 0 else: dp[i][j][2] = dp[i + 1][j][2] + 1 if i + 1 < n else 1 dp[i][j][3] = dp[i][j + 1][3] + 1 if j + 1 < n else 1 ans = 0 for i in range(n): for j in range(n): ans = max(ans, min(dp[i][j])) return ans1944. 队列中可以看到的人数1944. 队列中可以看到的人数class Solution: def canSeePersonsCount(self, heights: List[int]) -> List[int]: stack = [] ans = [0 for i in range(len(heights))] for i in range(len(heights) - 1, -1, -1): while stack and heights[i] > stack[-1]: ans[i] += 1 stack.pop() if stack: ans[i] += 1 stack.append(heights[i]) return ans2439. 最小化数组中的最大值2439. 最小化数组中的最大值使用二分法检查数class Solution: def minimizeArrayValue(self, nums: List[int]) -> int: right = 10 ** 9 left = 0 while left < right: mid = left + (right - left) // 2 if self.check(mid, nums): right = mid else: left = mid + 1 return left def check(self, k, nums): num = 0 for i in range(len(nums)): if nums[i] < k: num += k - nums[i] else: if num < (nums[i] - k): return False else: num -= nums[i] - k return TrueLCR 009. 乘积小于 K 的子数组LCR 009. 乘积小于 K 的子数组class Solution: def numSubarrayProductLessThanK(self, nums: List[int], k: int) -> int: cnt = 0 mul = 1 p0 = 0 for p1 in range(len(nums)): mul *= nums[p1] while p0 <= p1 and mul >= k: mul //= nums[p0] p0 += 1 cnt += p1 - p0 + 1 return cnt 966. 元音拼写检查器966. 元音拼写检查器class Solution: def spellchecker(self, wordlist: List[str], queries: List[str]) -> List[str]: words_set = set(wordlist) words_lower = {wordlist[i].lower(): wordlist[i] for i in range(len(wordlist) - 1, -1, -1)} words_vow = {"".join([c if c not in 'aeiou' else '*' for c in wordlist[i].lower()]): wordlist[i] for i in range(len(wordlist) - 1, -1, -1)} ans = [] for word in queries: if word in words_set: ans.append(word) continue word_lower = word.lower() if word_lower in words_lower: ans.append(words_lower[word_lower]) continue word_vow = "".join([c if c not in 'aeiou' else '*' for c in word_lower]) if word_vow in words_vow: ans.append(words_vow[word_vow]) continue ans.append('') return ans