轮询算法案例

本文目录导读：

1. 基础轮询（Round Robin）
2. 加权轮询（Weighted Round Robin）
3. 平滑加权轮询
4. 分布式系统中的轮询（带重试机制）
5. 实际问题：时间片轮转调度（CPU调度）
6. 实际场景：消息队列消费者轮询
7. 使用建议

我将为你提供几个不同场景下轮询算法的完整案例,从基础到进阶，包含代码实现和适用场景分析。

基础轮询（Round Robin）

最简单的轮询算法,适用于服务器负载均衡中最常见的场景。

class RoundRobinBalancer:
    def __init__(self, servers):
        self.servers = servers  # 服务器列表
        self.current_index = 0  # 当前索引
        self.lock = threading.Lock()  # 线程安全
    def get_server(self):
        """获取下一个服务器"""
        with self.lock:
            server = self.servers[self.current_index]
            self.current_index = (self.current_index + 1) % len(self.servers)
            return server
# 使用示例
servers = ["server1", "server2", "server3", "server4"]
balancer = RoundRobinBalancer(servers)
# 模拟10个请求
for request_id in range(10):
    server = balancer.get_server()
    print(f"请求 {request_id} -> 分配给 {server}")

输出结果：

请求 0 -> 分配给 server1
请求 1 -> 分配给 server2
请求 2 -> 分配给 server3
请求 3 -> 分配给 server4
请求 4 -> 分配给 server1
...依次循环

加权轮询（Weighted Round Robin）

考虑到服务器性能差异,高性能服务器承受更多请求。

class WeightedRoundRobin:
    def __init__(self, servers_with_weights):
        # 输入格式: [("server1", 5), ("server2", 3), ("server3", 2)]
        self.servers = []
        for server, weight in servers_with_weights:
            self.servers.extend([server] * weight)  # 按权重扩展列表
        self.current_index = 0
        self.total_requests = 0
    def get_server(self):
        """获取下一个服务器（按权重分配）"""
        server = self.servers[self.current_index % len(self.servers)]
        self.current_index += 1
        return server
# 使用示例
servers_with_weights = [
    ("高性能服务器", 5),
    ("中性能服务器", 3), 
    ("低性能服务器", 2)
]
balancer = WeightedRoundRobin(servers_with_weights)
# 模拟请求
for i in range(10):
    server = balancer.get_server()
    print(f"请求 {i}: {server}")

输出结果：

请求 0: 高性能服务器
请求 1: 高性能服务器
请求 2: 高性能服务器
请求 3: 高性能服务器
请求 4: 高性能服务器
请求 5: 中性能服务器
请求 6: 中性能服务器
请求 7: 中性能服务器
请求 8: 低性能服务器
请求 9: 低性能服务器

平滑加权轮询

Nginx使用的平滑加权轮询算法,避免权重大的服务器被连续请求。

class SmoothWeightedRoundRobin:
    def __init__(self, servers_with_weights):
        # 服务器配置: [("server1", weight), ...]
        self.servers = []
        self.current_weights = {}
        for server, weight in servers_with_weights:
            self.servers.append({"name": server, "weight": weight})
            self.current_weights[server] = 0
    def get_server(self):
        """平滑加权轮询"""
        total_weight = sum(s["weight"] for s in self.servers)
        # 选择当前权重最高的服务器
        selected = None
        max_weight = 0
        for server in self.servers:
            name = server["name"]
            weight = server["weight"]
            # 当前权重加上自身权重
            self.current_weights[name] += weight
            # 找出当前权重最大的
            if self.current_weights[name] > max_weight:
                max_weight = self.current_weights[name]
                selected = name
        # 被选中的服务器减去总权重
        if selected:
            self.current_weights[selected] -= total_weight
        return selected
# 使用示例
servers = [
    ("A", 5),  # A服务器权重5
    ("B", 1),  # B服务器权重1
    ("C", 1)   # C服务器权重1
]
balancer = SmoothWeightedRoundRobin(servers)
# 模拟10个请求
for i in range(10):
    server = balancer.get_server()
    print(f"请求 {i}: 选择服务器 {server}")

输出结果：

请求 0: A
请求 1: A
请求 2: B
请求 3: A
请求 4: A
请求 5: C
请求 6: A
请求 7: A
请求 8: B
请求 9: A

分布式系统中的轮询（带重试机制）

import time
import random
class RobustRoundRobin:
    def __init__(self, nodes):
        self.nodes = nodes  # 节点列表
        self.current_index = 0
        self.failed_nodes = set()  # 故障节点
        self.max_retries = 3  # 最大重试次数
    def get_healthy_node(self):
        """获取健康节点，跳过故障节点"""
        attempts = 0
        node_count = len(self.nodes)
        while attempts < node_count:
            node = self.nodes[self.current_index % node_count]
            self.current_index += 1
            if node not in self.failed_nodes:
                return node
            attempts += 1
        # 所有节点都故障，返回None
        return None
    def process_request(self, request_id):
        """处理请求，带重试机制"""
        retry_count = 0
        while retry_count < self.max_retries:
            node = self.get_healthy_node()
            if node is None:
                print(f"请求 {request_id}: 所有节点都不可用")
                return False
            try:
                # 模拟请求处理
                success = self.call_node(node, request_id)
                if success:
                    print(f"请求 {request_id}: {node} 处理成功")
                    return True
                else:
                    # 节点处理失败，标记为故障
                    self.failed_nodes.add(node)
                    print(f"请求 {request_id}: {node} 处理失败")
                    retry_count += 1
            except Exception as e:
                print(f"请求 {request_id}: {node} 异常 - {e}")
                self.failed_nodes.add(node)
                retry_count += 1
        print(f"请求 {request_id}: 重试{retry_count}次后失败")
        return False
    def call_node(self, node, request_id):
        """模拟调用节点"""
        # 模拟随机失败
        return random.random() > 0.3  # 70%成功率
# 使用示例
nodes = ["Node-1", "Node-2", "Node-3", "Node-4", "Node-5"]
balancer = RobustRoundRobin(nodes)
# 模拟请求
for request_id in range(1, 6):
    balancer.process_request(request_id)

实际问题：时间片轮转调度（CPU调度）

class Process:
    def __init__(self, pid, burst_time):
        self.pid = pid
        self.burst_time = burst_time  # 需要的CPU时间
        self.remaining_time = burst_time
        self.wait_time = 0
        self.turnaround_time = 0
class TimeSliceScheduler:
    def __init__(self, time_quantum):
        self.time_quantum = time_quantum  # 时间片大小
        self.process_queue = []
        self.completed_processes = []
    def add_process(self, process):
        self.process_queue.append(process)
    def run(self):
        """运行调度器"""
        current_time = 0
        while self.process_queue:
            process = self.process_queue.pop(0)
            # 执行时间片
            if process.remaining_time > self.time_quantum:
                print(f"时间 {current_time}-{current_time+self.time_quantum}: "
                      f"进程 {process.pid} 执行 {self.time_quantum}ms")
                process.remaining_time -= self.time_quantum
                current_time += self.time_quantum
                # 未完成，放回队列尾部
                process.wait_time += 0  # 等待时间在后续更新
                self.process_queue.append(process)
            else:
                # 进程完成
                print(f"时间 {current_time}-{current_time+process.remaining_time}: "
                      f"进程 {process.pid} 执行 {process.remaining_time}ms (完成)")
                current_time += process.remaining_time
                process.remaining_time = 0
                process.turnaround_time = current_time
                self.completed_processes.append(process)
        self.print_stats()
    def print_stats(self):
        """打印统计信息"""
        print("\n=== 调度统计 ===")
        for process in self.completed_processes:
            total_wait = process.turnaround_time - process.burst_time
            print(f"进程 {process.pid}: 执行时间={process.burst_time}ms, "
                  f"完成时间={process.turnaround_time}ms, "
                  f"等待时间={total_wait}ms")
# 使用示例
scheduler = TimeSliceScheduler(time_quantum=4)
# 创建进程
processes = [
    Process(1, 10),  # PID 1，需要10ms
    Process(2, 5),   # PID 2，需要5ms  
    Process(3, 7),   # PID 3，需要7ms
]
for p in processes:
    scheduler.add_process(p)
scheduler.run()

输出结果：

时间 0-4: 进程 1 执行 4ms
时间 4-8: 进程 2 执行 4ms
时间 8-11: 进程 3 执行 3ms
时间 11-15: 进程 1 执行 4ms
时间 15-17: 进程 3 执行 2ms
时间 17-23: 进程 2 执行 1ms
=== 调度统计 ===
进程 2: 执行时间=5ms, 完成时间=23ms, 等待时间=18ms
进程 3: 执行时间=7ms, 完成时间=17ms, 等待时间=10ms
进程 1: 执行时间=10ms, 完成时间=19ms, 等待时间=9ms

实际场景：消息队列消费者轮询

import threading
import time
from queue import Queue
class MessageQueueConsumer:
    def __init__(self, consumer_count=3):
        self.queue = Queue()
        self.consumers = [f"Consumer-{i}" for i in range(consumer_count)]
        self.current_index = 0
        self.message_count = 0
    def add_message(self, message):
        """添加消息到队列"""
        self.queue.put(message)
        print(f"新增消息: {message}")
    def process_messages(self):
        """处理消息 - 轮询分配"""
        while not self.queue.empty():
            # 获取消息
            message = self.queue.get()
            # 轮询选择消费者
            consumer = self.consumers[self.current_index % len(self.consumers)]
            self.current_index += 1
            # 模拟处理
            print(f"[{time.strftime('%H:%M:%S')}] {consumer} 处理消息: {message}")
            time.sleep(0.5)  # 模拟处理时间
            self.queue.task_done()
# 使用示例
consumer = MessageQueueConsumer(3)
# 添加一些消息
for i in range(1, 8):
    consumer.add_message(f"消息{i}")
print("\n开始处理...")
consumer.process_messages()

输出结果：

新增消息: 消息1
新增消息: 消息2
新增消息: 消息3
新增消息: 消息4
新增消息: 消息5
新增消息: 消息6
新增消息: 消息7
开始处理...
[10:30:25] Consumer-0 处理消息: 消息1
[10:30:26] Consumer-1 处理消息: 消息2
[10:30:26] Consumer-2 处理消息: 消息3
[10:30:27] Consumer-0 处理消息: 消息4
[10:30:27] Consumer-1 处理消息: 消息5
[10:30:28] Consumer-2 处理消息: 消息6
[10:30:28] Consumer-0 处理消息: 消息7

使用建议

1. 最简单场景：使用基础轮询即可
2. 服务器性能差异大：使用加权轮询
3. 避免请求集中：使用平滑加权轮询（适合Nginx）
4. 系统可靠性要求高：加入健康检查和重试机制
5. CPU调度：使用时间片轮转算法

轮询算法核心优点是简单、公平、无状态，非常适合负载均衡等场景，根据具体需求选择合适的变体即可。