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配置项时它是无效的。 } }
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"))
1711. 大餐计数1711. 大餐计数from typing import List from collections import Counter import math class Solution: def countPairs(self, deliciousness: List[int]) -> int: min_deliciousness = min(deliciousness) if min_deliciousness <= 0: min_deliciousness = 0.5 max_deliciousness = max(deliciousness) count = Counter(deliciousness) ans = 0 wait_list = [2 ** i for i in range(int(math.log(min_deliciousness * 2, 2)), int(math.log(max_deliciousness * 2, 2)) + 1)] used = set() for k, v in count.items(): for i in range(len(wait_list)): if wait_list[i] - k in count: if wait_list[i] - k == k and v > 1: ans += (1 + v - 1) * (v - 1) // 2 ans = ans % (10 ** 9 + 7) elif wait_list[i] - k == k: pass else: if wait_list[i] - k not in used: ans += v * count[wait_list[i] - k] ans = ans % (10 ** 9 + 7) used.add(k) return ans % (10 ** 9 + 7)1702. 修改后的最大二进制字符串1702. 修改后的最大二进制字符串class Solution: def maximumBinaryString(self, binary: str) -> str: left = 0 binary = list(binary) while left < len(binary) and binary[left] == '1': left += 1 right = left + 1 while right < len(binary): if binary[right] == '1': right += 1 continue if right - left == 1: binary[left] = '1' else: binary[left] = '1' binary[right] = '1' binary[left + 1] = '0' left += 1 right += 1 return ''.join(binary)133. 克隆图133. 克隆图class Node: def __init__(self, val=0, neighbors=None): self.val = val self.neighbors = neighbors if neighbors is not None else [] from typing import Optional class Solution: def __init__(self): self.created = {} self.used = set() def cloneGraph(self, node: Optional['Node']) -> Optional['Node']: if node is None: return None val = node.val self.dfs(node) return self.created[val] def dfs(self, node: Optional['Node']): if node.val in self.used: return if node.val not in self.created: new_node = Node(node.val) self.created[node.val] = new_node else: new_node = self.created[node.val] self.used.add(node.val) for i in range(len(node.neighbors)): if node.neighbors[i].val in self.created: new_node.neighbors.append(self.created[node.neighbors[i].val]) else: tmp_node = Node(node.neighbors[i].val) self.created[tmp_node.val] = tmp_node new_node.neighbors.append(tmp_node) self.dfs(node.neighbors[i]) 1991. 找到数组的中间位置1991. 找到数组的中间位置class Solution: def findMiddleIndex(self, nums: List[int]) -> int: right = sum(nums) left = 0 for i in range(len(nums)): right -= nums[i] if left == right: return i left += nums[i] return -1994. 腐烂的橘子994. 腐烂的橘子class Solution: def __init__(self): self.time = 0 def orangesRotting(self, grid: List[List[int]]) -> int: rotting_oranges = set() normal_oranges = set() for i in range(len(grid)): for j in range(len(grid[i])): if grid[i][j] == 2: rotting_oranges.add((i, j)) if grid[i][j] == 1: normal_oranges.add((i, j)) if not normal_oranges: return 0 while rotting_oranges: rotting_oranges_bak = set(rotting_oranges) for (x, y) in rotting_oranges_bak: if x - 1 >= 0 and grid[x - 1][y] == 1: grid[x - 1][y] = 2 rotting_oranges.add((x - 1, y)) if x + 1 < len(grid) and grid[x + 1][y] == 1: grid[x + 1][y] = 2 rotting_oranges.add((x + 1, y)) if y - 1 >= 0 and grid[x][y - 1] == 1: grid[x][y - 1] = 2 rotting_oranges.add((x, y - 1)) if y + 1 < len(grid[x]) and grid[x][y + 1] == 1: grid[x][y + 1] = 2 rotting_oranges.add((x, y + 1)) rotting_oranges = rotting_oranges - rotting_oranges_bak normal_oranges = normal_oranges - rotting_oranges self.time += 1 return self.time - 1 if not normal_oranges else -1
2260. 必须拿起的最小连续卡牌数2260. 必须拿起的最小连续卡牌数class Solution: def minimumCardPickup(self, cards: List[int]) -> int: res = {} mark = {cards[k]: False for k in range(len(cards))} for i in range(len(cards)): if cards[i] in res: if res[cards[i]][1] - res[cards[i]][0] > i - res[cards[i]][2] or res[cards[i]][1] == res[cards[i]][0]: res[cards[i]] = [res[cards[i]][2], i, i] mark[cards[i]] = True res[cards[i]][2] = i else: res[cards[i]] = [i, i, i] ans = float('inf') for k, v in res.items(): if v[0] == v[1]: continue if mark[k] and ans > v[1] - v[0] + 1: ans = v[1] - v[0] + 1 return ans if ans != float('inf') else -11893. 检查是否区域内所有整数都被覆盖1893. 检查是否区域内所有整数都被覆盖class Solution: def isCovered(self, ranges: List[List[int]], left: int, right: int) -> bool: ranges.sort(key=lambda x: x[0]) father = [ranges[i][0] for i in range(len(ranges))] for i in range(1, len(ranges)): if ranges[i][0] < ranges[i - 1][1] and ranges[i][1] < ranges[i - 1][1]: ranges[i][1] = ranges[i - 1][1] for i in range(1, len(ranges)): if ranges[i][0] - ranges[i - 1][1] <= 1: father[i] = father[i - 1] left_father = None right_father = None for i in range(len(ranges)): if ranges[i][0] <= left <= ranges[i][1] and not left_father: left_father = father[i] if ranges[i][0] <= right <= ranges[i][1] and not right_father: right_father = father[i] if left_father and right_father: break if left_father is None and right_father is None: return False else: return left_father == right_father2269. 找到一个数字的 K 美丽值2269. 找到一个数字的 K 美丽值class Solution: def divisorSubstrings(self, num: int, k: int) -> int: num = str(num) ans = 0 for i in range(len(num) - k + 1): sub_num = int(num[i:i+k]) if sub_num == 0: continue if int(num) % sub_num == 0: ans += 1 return ans1779. 找到最近的有相同 X 或 Y 坐标的点1779. 找到最近的有相同 X 或 Y 坐标的点class Solution: def nearestValidPoint(self, x: int, y: int, points: List[List[int]]) -> int: ans = 0 distance = float('inf') for i in range(len(points)): if points[i][0] == x or points[i][1] == y: dis = abs(points[i][0] - x) + abs(points[i][1] - y) if dis < distance: distance = dis ans = i return ans if distance != float('inf') else -12233. K 次增加后的最大乘积2233. K 次增加后的最大乘积贪心import heapq class Solution: def maximumProduct(self, nums: List[int], k: int) -> int: heapq.heapify(nums) for i in range(k): heapq.heapreplace(nums, nums[0] + 1) ans = 1 for i in range(len(nums)): ans *= nums[i] ans %= (10 ** 9 + 7) return ans
VHDVHD 结构是一种虚拟磁盘的实现方式，即通过文件来模拟物理磁盘的方式来存储数据。VHD结构有2种实现方式：固定方式和动态方式。本文只讨论固定方式。固定方式就是用真实大小的文件模拟同样大小的一个虚拟磁盘。VHD 头部VHD和其他文件不同，VHD其信息放在尾部，也就是最后一个扇区中。我们应当首先读取最后一个扇区的内容。字段字节数说明cookie8标识，应为conectix，用于判断VHD是否有效features40： HD_NO_FEATURES1： HD_TEMPORARY2： HD_RESERVEDff_version4VHD版本data_offset8固定磁盘，总是0xFFFFFFFF。timestamp4VHD的创建时间，指2000年1月1日00:00:00至今的秒数crtr_app4创建者crtr_ver4创建版本crtr_os4创建者系统orig_size8创建的虚拟磁盘大小，这个大小指虚拟出来的磁盘的可用寻址空间。curr_size8用于vhd在线扩容后的最后大小表述，如果没有扩容，和orig_size相同。geometry4VHD的C/H/S结构参数柱面数2字节，磁头数1字节，扇区数1字节type40： HD_TYPE_NONE 2： HD_TYPE_FIXED 固定磁盘3： HD_TYPE_DYNAMIC 动态磁盘4： HD_TYPE_DIFF 差异磁盘checksum4本扇区的校验和uuid16磁盘标识，在差异磁盘中，决定了VHD间的主从关系。saved1是否在保存状态hidden1VDI是否隐藏VHD 写入VHD文件规定读写需要针对磁道，读写必须要在某个磁道的开头处，然后读取n个磁道（n*512字节）。如果你的数据不够512的倍数，那么写入的时候需要将数据凑到512字节的倍数，不够的部分补0。import datetime import logging class Vhd: def __init__(self, file_name): self.head = {} self.file_name = file_name self.f = open(self.file_name, 'rb+') self.head_info() def head_info(self): self.f.seek(-512, 2) self.head["cookie"] = self.f.read(8).decode() self.head["features"] = self.f.read(4) self.head["ff_version"] = self.f.read(4) self.head["data_offset"] = self.f.read(8) t = int.from_bytes(self.f.read(4)) t = datetime.datetime(2000, 1, 1, 8) + datetime.timedelta(seconds=t) self.head["timestamp"] = t.strftime("%Y-%m-%d %H:%M:%S") self.head["crtr_app"] = self.f.read(4).decode() self.head["crtr_ver"] = self.f.read(4) self.head["crtr_os"] = self.f.read(4) self.head["orig_size"] = int.from_bytes(self.f.read(8)) self.head["curr_size"] = int.from_bytes(self.f.read(8)) cylinder = int.from_bytes(self.f.read(2)) head = int.from_bytes(self.f.read(1)) sector = int.from_bytes(self.f.read(1)) self.head["geometry"] = (cylinder, head, sector) type_ = int.from_bytes(self.f.read(4)) if type_ == 0: type_ = "HD_TYPE_NONE" elif type_ == 2: type_ = "HD_TYPE_FIXED" elif type_ == 3: type_ = "HD_TYPE_DYNAMIC" elif type_ == 4: type_ = "HD_TYPE_DIFF" self.head["type"] = type_ self.head["checksum"] = self.f.read(4) self.head["uuid"] = self.f.read(16) self.head["saved"] = bool(int.from_bytes(self.f.read(1))) self.head["hidden"] = bool(int.from_bytes(self.f.read(1))) def __str__(self): return str(self.head) def write(self, bin_file, *, lba=None, cylinder=None, head=None, sector=None): if lba is None: lba = cylinder * self.head['geometry'][1] * self.head['geometry'][2] + head * self.head['geometry'][ 2] + sector - 1 start_pos = lba * 512 with open(bin_file, 'rb') as f: self.f.seek(-512, 2) safe = self.f.tell() f.seek(0, 2) write_file_length = f.tell() if start_pos + write_file_length > safe: raise Exception('No Enough Space') self.f.seek(start_pos, 0) f.seek(0, 0) while True: data = f.read(512) if not data: break self.f.write(data) logging.info(f'file write successfully') def __del__(self): self.f.close()
家庭1. 研究家庭的主要方法恩格斯在开篇提到摩尔根的发现：在易洛魁人中间盛行的，是一种双方可以轻易解除的个体婚姻，摩尔根称之为“对偶制家庭”。这种家庭形式中，父母、子女、兄弟、姐妹之间的关系非常明确，但是，这些称呼的实际使用，却与此矛盾。易洛魁人的男子，不仅把自己亲生的子女称为自己的儿子和女儿，而且把他兄弟的子女也称为自己的儿子和女儿，而他们都称他为父亲，他把自己姊妹的子女则称为自己的外甥和外甥女，他们称他为舅父。同样，易洛魁人的女子，把自己姊妹的子女和她自己亲生的子女一概都称为自己的儿子和女儿，而他们都称她为母亲，她把自己兄弟的子女称为自己的内侄和内侄女，她自己被称为他们的姑母。这样的亲属关系同亲属制度矛盾，是怎样产生的呢？摩尔根在《古代社会》中说：“家庭是一个能动的要素；它从来不是静止不动的，而是随着社会从较低阶段向较高阶段的发展，从低级的形式如进到较高的形式。反之，亲属制度却是被动的；它只是把家庭经过一个长久时期所发生的进步记录下来，并且只是在家庭已经根本变化了的时候，它才发生根本的变化。”这句话也就是说：亲属制度是对亲属关系的反映。当一种家庭形式确立起来，相应的亲属制度也随之确立，但是在家庭形式变动过程中，亲属制度可能还没那么迅速地随之发生变化，还以习惯的方式继续存在着。但是有一点是明确的，没有一定的家庭形式也就不可能形成与之相应的亲属制度，反之，由亲属制度追根溯源必然能在现实的历史发展过程找寻到与之相应的家庭关系。于是，我们可以得出这样一个结论：如果现行家庭形式中产生的亲属关系，是同亲属制度相矛盾的，那么在这之前必然存在一种家庭形式，其产生的亲属关系是与现行的亲属制度相适应的，由现行的亲属制度追根溯源必然能在现实的历史发展过程中找寻到与之相应的家庭关系。2.原始的家庭形式​恩格斯认为原始的家庭形式就是“群婚”，即“整群的男子与整群的女子互为所有”。产生群婚这种家庭形式存在两个前提：一是以群的联合力量和集体行动来弥补个体自卫能力的不足。二是成年雄者的相互宽容，没有忌妒。​关于第二点，我们不得不提到高等动物中群和家庭的关系。恩格斯认为，较高等级的脊椎动物与人类在家庭和群的关系有重要的差别：​较高等级的脊椎动物由于雄性存在本能性的嫉妒，使其联结又限制着动物家庭的发展，又同样使其的家庭跟群对立起来，在家庭紧密结合的地方，群只是一种稀有的例外。反之，在自由的性关系或多偶制盛行的地方，群差不多是自动地形成的……为了使群能够组成，家庭的纽结必然要放松，个体必然要重新自由。因此我们可以得从这样一个结论：高等动物的群和家庭并不是互相补充，而是互相对立的。雄者的嫉妒使高等动物的群基本不可能产生。"而成年雄者的相互宽容，没有忌妒，则是形成较大的持久的集团的首要条件。"也就是说，原始状态下的人类在家庭和群之间的选择是明确的，为了要在凶险的自然界中维持生存，就必须要选择后者，不能因为个体的家庭而抛弃共居生活的群。​在此基础上，群婚便自然而然的产生出来了。与这个阶段相适应的性关系被恩格斯称之为杂乱的性关系，即不排除兄弟姐妹之间，父母与子女之间的性关系。人类之后的家庭形式，便在此基础上发展起来。3.几种家庭形式血缘家庭婚姻集团按辈分划分，即在家庭范围内，一群直系或旁系的兄弟姊妹互相通婚，但婚姻关系基本上排除祖辈与孙辈、父母辈与子女辈的婚配。这种家庭的典型形式是一对配偶的子孙中，每一代都互为兄弟姊妹，也互为夫妻。在亲属称谓上无父系和母系的区别，祖父与外祖父、伯叔父与舅父、姑母与姨母、舅母与母亲等都使用相同称呼。即一对配偶的子孙中每一代都互为兄弟姊妹，也互为夫妻。目前血缘家庭已经绝迹，但在夏威夷的亲属制度揭示了它的存在。普那路亚家庭血缘家庭排除了父母与子女之间的性关系，而普那路亚家庭更进一步，排除了兄弟姊妹之间的性关系。按照夏威夷的习俗，若干数目的姊妹——同胞的或血统较远的即从（表）姊妹，再从（表）姊妹或更远一些的姊妹——是她们共同丈夫们的共同的妻子，但是在这些共同丈夫之中，排除了她们的兄弟；这些丈夫彼此已不再互称兄弟，他们也不再必须是兄弟了，而是互称普那路亚，即亲密的同伴，即所谓Associé。同样，一列兄弟——同胞的或血统较远的——则跟若干数目的女子（只要不是自己的姊妹）共同结婚，这些女子也互称普那路亚。普那路亚家庭是群婚的最高发展阶段，普那路亚家庭的出现对人类历史的发展造成了巨大的影响，氏族就是在普那路亚家庭的基础上直接发展而来的。氏族就是由这一进步直接引起的，而且远远超过了最初的目的，它构成地球上即使不是所有的也是大多数野蛮民族的社会制度的基础。所有的群婚家庭中，只有孩子的母亲是能够被确定的，孩子的父亲是不确定的。母亲能够把自己亲生的子女同其他一切并非自己亲生的子女区别开来，即使母亲也称共同家庭中的其他孩子为自己的子女。也就是说，只要存在着群婚，那么世系就只能从母亲的方面来确定。现在我们面前有这样一个集团，假设集团中的女性有一个共同的女始祖，那么她们的每一代都互为姊妹，每一代姊妹的丈夫却不能是她们的兄弟了，即她们的丈夫是不能出自这个女始祖的，是不能属于这个集团的。又由于世系只能从母亲的方面来确定，所以她们子女又属于这个集团。由于在普那路亚家庭中，一切兄弟和姊妹间，甚至母方最远的旁系亲属间的性关系都不被允许，上述的集团便转化为氏族。对偶制家庭由于婚姻禁规日益错综复杂，群婚就越来越不可能；群婚就被对偶制家庭排挤了。在这一阶段上，一个男子和一个女子共同生活；这种状态，在群婚制度下或者更早的时候就已经发生了，一个男子在许多妻子中有一个主妻，而他对于这个女子来说是她的许多丈夫中的最主要的丈夫。这种婚姻关系很容易由任何一方解除。解除婚姻关系之后，子女仍同往常一样划归母亲一方。相比于以往的婚姻制度，对偶婚更有利于人类种群的繁衍。由此可见，原始历史上家庭的发展，就在于不断缩小最初包括整个部落并在内部盛行两性共同婚姻的那个范围。由于次第排斥亲属通婚（起初是血统较近的，后来是血统越来越远的亲属，最后甚至是仅有姻亲关系的），任何群婚形式终于在实际上成为不可能的了，结果，只剩下一对暂时松散地结合的配偶，即一旦解体整个婚姻就终止的分子。对偶制家庭还无法实行以家庭为单位的私有制。因此，对偶制家庭存在的时代仍然以共产制家户经济为主。由于大多数或全体妇女都属于同一氏族，而男子则来自不同的氏族，且子女仍同往常一样划归母亲一方则意味着女性仍在家中占有统治地位。决定两性间的分工的原因，是同决定妇女社会地位的原因完全不同的。有些民族的妇女所做的工作比我们所设想的要多得多，这些民族比我们欧洲人常常对妇女怀着更多的真正尊敬。外表上受尊敬的、脱离一切实际劳动的文明时代的贵妇人，比起野蛮时代辛苦劳动的妇女来，其社会地位是无比低下的；后者在本民族中被看做真正的贵妇人，而就其地位的性质说来，她们也确是如此。在群婚向个体婚制转变过程中，妇女是绝对的主导。古代遗传下来的两性间的关系，越是随着经济生活条件的发展，从而随着古代共产制的解体和人口密度的增大，而失去森林原始生活的素朴性质，就必然越使妇女感到屈辱和压抑；妇女也就必然越迫切地要求取得保持贞操的权利，取得暂时地或长久地只同一个男子结婚的权利作为解救的办法。专偶制家庭​随着财富的逐渐累积，财富的归属问题自然也成为摆在原始人类面前的重要问题。原始社会末期，随着生产力的逐步提升，人类聚居之处逐渐形成适宜私有制的土壤，古代共产制解体后，财产的继承问题迫使家庭形式发生改变。​除了生身的母亲之外，对偶婚确定了确实的生身父亲。由于分工和财产的私有化，按照当时的家庭分工，丈夫的责任是获取食物以及劳动工具，从而也拥有劳动工具的所有权。这些劳动工具成为他的私有财产，即便是离婚了它们也要随父亲而去。因此，按照当时的社会习惯，男子既是家畜的所有者也是奴隶的所有者。然而，在母权制下，孩子与其父亲属于不同的氏族，因此父亲的这些财产要归其自身所在的氏族所有，而他的亲生子女则不具有继承权。这样，财产的继承问题随着对偶制家庭的建立第一次成为重要问题。随着财富的增加，财富便一方面使丈夫在家庭中占据比妻子更重要的地位；另一方面，又产生了利用这个增强了的地位来废除传统的继承制度使之有利于子女的原动力​由于这种继承制度问题根源于母权制，因此必须废除母权制。人类所经历过的最深刻的革命之一——并不需要侵害到任何一个活着的氏族成员。氏族的全体成员都仍然能够和以前一样。只要有一个简单的决定，规定以后氏族男性成员的子女应该留在本氏族内，而女性成员的子女应该离开本氏族，转到他们父亲的氏族中去就行了。​这场革命的深远影响首先就体现在它对女性地位的极大冲击。男子在家庭中获得了统治地位，而女子的地位则显著下降，甚至成为丈夫的工具。丈夫在家中也掌握了权柄，而妻子则被贬低，被奴役，变成丈夫淫欲的奴隶，变成单纯的生孩子的工具了。​专偶制这种家庭形式建立在丈夫的统治之上，其目的是确定父亲与子女之间确凿无疑的关系。而这又是为了使父亲的财产能够为自己亲生的子女继承出发来考虑的。专偶制家庭和对偶制家庭的不同在于，“婚姻关系要牢固得多”，“不能由双方任意解除”。但是，丈夫却享有解除婚姻乃至对婚姻不忠的权利。并且，随着社会的发展，这种权利也被越来越广泛地行使。在母权制社会中母亲所享有的真正的尊敬已经一去不复返了。正是奴隶制与专偶制的并存，正是完全受男子支配的年轻美貌的女奴隶的存在，使专偶制从一开始就具有了它的特殊的性质，使它成了只是对妇女而不是对男子的专偶制。​专偶制家庭的出现从根本上来看以生产的发展和私有制的出现为前提，财产的私有化及财产在父子间延续是其产生的根本动力。另一方面，专偶制的出现也是阶级分化、阶级对立的产物。个体婚制在历史上决不是作为男女之间的和好而出现的，更不是作为这种和好的最高形式而出现的。它是作为整个史前时代所没有的两性冲突—特别是男性对女性的奴役之宣告而出现的。​个体婚制虽然是一个伟大的历史进步，但它也是伴随着奴隶制和私有制而来的。从历史发展进程来看，这种家庭形式至今也仍然是最普遍的家庭形式。它之所以具有持久性和稳定性，就在于维持其存续的历史条件即私有制仍然存在并且必然仍将存在一段时间。换言之，只要私有制尚且存在，个体婚制、专偶制也必然会持续存在下去。可见，个体婚制在历史上决不是作为男女之间的和好而出现的，更不是作为这种和好的最高形式而出现的。恰好相反。它是作为女性被男性奴役，作为整个史前时代所未有的两性冲突的宣告而出现的。​现代资产阶级的婚姻总是权衡利弊的婚姻。在天主教国家中，婚姻由父母做主，其结果是专偶制所固有的矛盾以最尖锐的方式表现出来：丈夫大肆淫游，而妻子则大肆通奸。在新教国家中会有些细微差别。新教的伪善而不够严格使得婚姻的缔结可以允许一定程度的爱恋。虽然如此，婚姻的本质也无法从根本上得到改变。这种权衡利害的婚姻，在这两种场合都往往变为最粗鄙的卖淫——有时是双方，而更常见的是妻子。妻子和普通娼妓的不同之处，只在于她不是像雇佣女工做计件工作那样出租自己的身体，而是把身体一次永远出卖为奴隶。根据我们对古代最文明、最发达的民族所能作的考察，专偶制的起源就是如此。它绝不是个人性爱的结果，它同个人性爱绝对没有关系，因为婚姻和以前一样仍然是权衡利害的婚姻。专偶制是不以自然条件为基础，而以经济条件为基础，即以私有制对原始的自然产生的公有制的胜利为基础的第一个家庭形式。未来婚姻​无产阶级当中才会产生真正的爱情和以爱情为基础的婚姻。因为在无产阶级当中，古典的专偶制婚姻得以存在的基础都不复存在了。专偶制以私有制的出现为前提，并且只有在私有制条件下才会出现如何将父亲的财产更合理合法地转移给自己的子女的问题。这些前提在无产阶级当中都丧失其价值。无产阶级丧失任何财产，劳动力是其剩下的唯一财富。​专偶制家庭束缚了女性的活动自由。但是，在资产阶级社会，无产阶级妇女也不得不被大工业卷入生产环节当中，他们和男人一样也是家庭财富的直接创造者，甚至在某些时期，妇女比男子更能适合某些工作，因此她们在特定时期能够担当家庭的主要劳动力。这样男子在女性面前的所谓统治的基础业已丧失殆尽。这样，专偶制存在的一切赖以维系的动力和基础在无产阶级家庭中均已消失。在包括许多夫妇和他们的子女的古代共产制家户经济中，由妇女料理家务，正如由男子获得食物一样，都是一种公共的、为社会所必需的事业。随着家长制家庭，尤其是随着专偶制个体家庭的产生，情况就改变了。料理家务失去了它的公共的性质。它与社会不再相干了。它变成了一种私人的服务；妻子成为主要的家庭女仆，被排斥在社会生产之外。只有现代的大工业，才又给妇女——只是给无产阶级的妇女——开辟了参加社会生产的途径。妇女解放的第一个先决条件就是一切女性重新回到公共的事业中去；而要达到这一点，又要求消除个体家庭作为社会的经济单位的属性。这样一来，男子的地位无论如何要发生很大的变化。而妇女的地位，一切妇女的地位也要发生很大的转变。随着生产资料转归公有，个体家庭就不再是社会的经济单位了。私人的家务变为社会的事业。孩子的抚养和教育成为公共的事情；社会同等地关怀一切儿童，无论是婚生的还是非婚生的。
Python 对象基础Python 整数对象Python 字符串对象Python 列表对象Python 字典对象Python 字节码对象Python 栈帧对象Python 简单表达式Python 复杂表达式Python 控制流程