UR机器人Socket通信实战：从零构建稳定可靠的数据通道

最近在几个自动化集成项目里，我频繁地和UR机器人打交道。说实话，刚开始接触UR的Socket通信时，踩的坑比预想的多得多——连接时断时续、指令执行延迟、偶尔还会遇到机器人“装聋作哑”的情况。经过几个月的实战摸索，我逐渐整理出了一套相对完整的避坑指南。这篇文章不是官方文档的复述，而是基于真实项目经验，从环境配置到调试优化的全流程解析，希望能帮你少走弯路。

无论是用Python快速原型开发，还是用C++构建高性能控制系统，UR机器人的Socket通信都是绕不开的核心技术。但很多人只关注“怎么连上”，却忽略了“怎么连得稳”、“怎么连得快”这些更实际的问题。今天我们就深入聊聊这些细节。

1. 环境配置：不只是IP和端口那么简单

很多人以为UR机器人的Socket通信配置就是填个IP地址和端口号，但实际上，前期的环境准备决定了后续通信的稳定性和可靠性。我见过太多项目因为基础配置不当，导致后期调试困难重重。

1.1 网络环境搭建的常见误区

UR机器人的控制器默认使用192.168.1.x网段，但很多开发环境并不在这个网段。直接修改机器人IP看似简单，但在实际生产环境中可能会引发连锁问题。

网络配置的黄金法则：

• 如果只是测试环境，可以修改机器人IP适配你的网络
• 在生产环境中，建议使用独立的网络交换机，将机器人和控制PC置于同一子网
• 避免使用无线网络进行实时控制通信，有线连接的稳定性要高几个数量级

注意：UR机器人的30002端口用于实时数据反馈，30003端口用于脚本指令发送。这两个端口的用途不同，不要混用。

1.2 开发环境的选择与配置

根据我的经验，不同的开发语言在UR通信中有各自的优势和适用场景：

开发语言	适用场景	性能特点	学习曲线
Python	快速原型、测试脚本、数据分析	开发快，但实时性一般	平缓，适合初学者
C++	高性能控制、实时系统、生产环境	延迟低，资源占用少	较陡，需要一定基础
C#	Windows平台集成、上位机开发	与.NET生态集成好	中等

如果你刚开始接触UR机器人，我建议从Python入手。不是因为它最好，而是因为它能让你快速验证想法，看到结果。等基本流程跑通后，再根据实际需求考虑是否迁移到C++。

Python环境配置其实很简单，但有几个细节容易忽略：

# 正确的导入方式
import socket
import time
import struct
import threading

# 常见错误：忘记设置超时时间
# 不设置超时会导致程序在连接失败时无限等待
socket.setdefaulttimeout(5.0)  # 设置全局默认超时

C++的环境配置相对复杂，特别是涉及到跨平台时。如果你用Qt，qsocket库是个不错的选择；如果追求极致性能，可以考虑直接使用系统原生的socket API。

2. 连接建立：不仅仅是connect()调用

建立Socket连接听起来简单，但UR机器人的连接建立有几个特殊之处。很多人在这里遇到的第一个坑就是：连接成功了，但发送指令没反应。

2.1 连接建立的正确姿势

UR机器人的Socket服务在30003端口上是一个持续监听的服务。但仅仅建立TCP连接是不够的，你还需要了解UR的通信协议特性。

def connect_to_ur(ip="192.168.1.100", port=30003, timeout=5):
    """
    建立与UR机器人的Socket连接
    返回：socket对象或None（连接失败时）
    """
    try:
        # 创建socket对象
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        
        # 设置超时（重要！）
        sock.settimeout(timeout)
        
        # 建立连接
        sock.connect((ip, port))
        
        # 验证连接是否真正建立
        # UR机器人连接建立后会立即发送状态数据
        # 我们可以尝试接收一点数据来验证
        sock.settimeout(1.0)  # 设置短超时进行验证
        try:
            data = sock.recv(1024)
            if data:
                print(f"连接成功，收到初始数据长度：{len(data)}字节")
        except socket.timeout:
            # 没有立即收到数据不一定代表连接失败
            # UR可能在特定状态下不发送初始数据
            print("连接建立，但未收到初始状态数据")
        
        # 恢复原来的超时设置
        sock.settimeout(timeout)
        
        return sock
        
    except Exception as e:
        print(f"连接失败：{e}")
        return None

这个连接函数有几个关键点：

1. 设置了连接超时：避免网络异常时程序卡死
2. 验证连接有效性：通过尝试接收数据确认连接真正建立
3. 异常处理完善：所有可能的异常都被捕获并处理

2.2 多连接管理的策略

在复杂的应用中，你可能需要同时管理多个Socket连接（比如同时连接多台机器人，或者同时使用30002和30003端口）。这时候就需要更精细的连接管理策略。

我常用的做法是创建一个连接管理器类：

class URConnectionManager:
    def __init__(self):
        self.connections = {}  # 存储所有活跃连接
        self.lock = threading.Lock()  # 线程锁，防止并发问题
        
    def add_connection(self, name, ip, port=30003):
        """添加一个新的连接"""
        with self.lock:
            if name in self.connections:
                print(f"连接 {name} 已存在")
                return False
                
            sock = connect_to_ur(ip, port)
            if sock:
                self.connections[name] = {
                    'socket': sock,
                    'ip': ip,
                    'port': port,
                    'last_activity': time.time()
                }
                return True
            return False
    
    def send_command(self, name, command):
        """通过指定连接发送指令"""
        with self.lock:
            if name not in self.connections:
                print(f"连接 {name} 不存在")
                return False
                
            conn = self.connections[name]
            try:
                conn['socket'].send(command.encode('utf-8'))
                conn['last_activity'] = time.time()
                return True
            except Exception as e:
                print(f"发送指令失败：{e}")
                # 标记连接为异常
                self._handle_broken_connection(name)
                return False
    
    def _handle_broken_connection(self, name):
        """处理断开的连接"""
        # 尝试重新连接
        # 记录错误日志
        # 通知上层应用
        pass

这种管理方式的好处是：

• 连接状态集中管理：所有连接的状态一目了然
• 自动重连机制：连接断开时可以自动尝试恢复
• 线程安全：支持多线程环境下的并发访问

3. 指令发送：UR脚本的语法与优化

发送指令到UR机器人不仅仅是把字符串扔过去那么简单。UR脚本语言有自己特定的语法和规则，理解这些规则能避免很多奇怪的问题。

3.1 UR脚本基础语法详解

UR机器人的脚本语言基于类似Python的语法，但有一些特殊的函数和约定。最常见的指令就是movej（关节运动）和movel（线性运动）。

基本运动指令格式：

# 关节运动到指定位置
movej([j1, j2, j3, j4, j5, j6], a=1.4, v=1.05, r=0)

# 线性运动到指定位置
movel(p[x, y, z, rx, ry, rz], a=1.2, v=0.25, r=0)

参数说明：

• a：加速度（rad/s² 或 m/s²）
• v：速度（rad/s 或 m/s）
• r：融合半径（m），0表示不融合
• t：时间（s），如果设置了t，会忽略a和v参数

3.2 指令发送的最佳实践

直接发送脚本字符串是最简单的方式，但在实际应用中，我们需要考虑更多因素：

def send_ur_script(sock, script_lines, wait_for_completion=True):
    """
    发送UR脚本指令
    
    参数：
    sock: socket连接对象
    script_lines: 脚本行列表
    wait_for_completion: 是否等待指令执行完成
    """
    # 将脚本行合并为完整脚本
    full_script = "\n".join(script_lines) + "\n"
    
    # 添加脚本结束标记（如果需要）
    if not full_script.strip().endswith("end"):
        full_script += "end\n"
    
    # 发送脚本
    try:
        sock.send(full_script.encode('utf-8'))
        print(f"已发送脚本：{full_script[:50]}...")
        
        # 如果需要等待执行完成
        if wait_for_completion:
            # UR机器人执行完成后会发送特定消息
            # 这里需要根据实际情况实现等待逻辑
            _wait_for_execution_complete(sock)
            
    except Exception as e:
        print(f"发送脚本失败：{e}")
        raise

def create_movej_command(joint_positions, a=1.4, v=1.05, r=0):
    """创建movej指令的辅助函数"""
    # 确保关节位置是6个值
    if len(joint_positions) != 6:
        raise ValueError("关节位置必须是6个值")
    
    # 格式化关节位置
    joints_str = ", ".join([str(p) for p in joint_positions])
    
    # 构建指令
    command = f"movej([{joints_str}], a={a}, v={v}, r={r})"
    return command

指令发送的几个重要注意事项：

1. 编码问题：UR机器人默认使用UTF-8编码，确保你的字符串正确编码
2. 换行符：每条指令通常以换行符结束，但完整的脚本块需要适当的结构
3. 执行确认：重要指令应该等待执行确认，而不是发送后就认为成功了

3.3 高级指令模式

对于复杂的运动序列，我推荐使用脚本函数的方式：

def create_complex_movement_script():
    """创建包含多个运动的复杂脚本"""
    script = [
        "def complex_move():",
        "  # 移动到安全位置",
        "  movej([0, -1.57, 0, -1.57, 0, 0], a=1.0, v=0.8)",
        "",
        "  # 接近目标点",
        "  movel(p[0.3, 0.2, 0.1, 3.14, 0, 0], a=0.5, v=0.1)",
        "",
        "  # 精细操作",
        "  movel(p[0.3, 0.2, 0.05, 3.14, 0, 0], a=0.2, v=0.05)",
        "",
        "  # 返回",
        "  movel(p[0.3, 0.2, 0.1, 3.14, 0, 0], a=0.5, v=0.1)",
        "  movej([0, -1.57, 0, -1.57, 0, 0], a=1.0, v=0.8)",
        "end"
    ]
    return script

这种方式的好处是：

• 原子性操作：整个运动序列作为一个整体发送和执行
• 减少通信次数：一次发送多个指令，减少网络延迟影响
• 更好的错误处理：可以在脚本中添加错误处理逻辑

4. 数据接收与状态监控

只发送指令是不够的，我们还需要接收机器人的状态反馈。UR机器人的30002端口专门用于实时数据反馈，这个功能在很多场景下都非常有用。

4.1 实时数据解析

UR机器人通过30002端口以125Hz的频率发送实时数据包。每个数据包包含机器人的各种状态信息：

def parse_ur_realtime_data(data):
    """
    解析UR实时数据包
    
    UR实时数据包格式（版本1.8）：
    前4字节：消息长度
    接下来是各种数据字段
    """
    if len(data) < 4:
        return None
    
    # 解析消息长度
    message_size = struct.unpack('!I', data[:4])[0]
    
    if len(data) < message_size:
        # 数据不完整
        return None
    
    # 解析各个字段
    offset = 4
    
    # 时间戳
    timestamp = struct.unpack('!d', data[offset:offset+8])[0]
    offset += 8
    
    # 关节位置（6个double）
    q_target = []
    for i in range(6):
        value = struct.unpack('!d', data[offset:offset+8])[0]
        q_target.append(value)
        offset += 8
    
    # 关节速度（6个double）
    qd_target = []
    for i in range(6):
        value = struct.unpack('!d', data[offset:offset+8])[0]
        qd_target.append(value)
        offset += 8
    
    # 继续解析其他字段...
    
    return {
        'timestamp': timestamp,
        'joint_positions': q_target,
        'joint_velocities': qd_target,
        # ... 其他字段
    }

提示：UR实时数据包的格式可能会因控制器版本不同而有所变化。在实际使用前，最好查阅对应版本的官方文档。

4.2 状态监控的实现

建立一个稳定的状态监控系统对于生产环境至关重要。下面是一个简单的监控实现：

class URStateMonitor:
    def __init__(self, ip="192.168.1.100", port=30002):
        self.ip = ip
        self.port = port
        self.socket = None
        self.monitoring = False
        self.callbacks = []
        
    def start_monitoring(self):
        """开始监控机器人状态"""
        if self.monitoring:
            return
            
        try:
            # 建立连接
            self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.socket.settimeout(1.0)
            self.socket.connect((self.ip, self.port))
            
            # 启动监控线程
            self.monitoring = True
            monitor_thread = threading.Thread(target=self._monitor_loop)
            monitor_thread.daemon = True
            monitor_thread.start()
            
            print("状态监控已启动")
            
        except Exception as e:
            print(f"启动监控失败：{e}")
            self.monitoring = False
    
    def _monitor_loop(self):
        """监控循环"""
        buffer = b""
        
        while self.monitoring:
            try:
                # 接收数据
                data = self.socket.recv(4096)
                if not data:
                    # 连接断开
                    break
                    
                buffer += data
                
                # 解析完整的数据包
                while len(buffer) >= 4:
                    # 检查消息长度
                    message_size = struct.unpack('!I', buffer[:4])[0]
                    
                    if len(buffer) >= message_size:
                        # 提取完整消息
                        message = buffer[:message_size]
                        buffer = buffer[message_size:]
                        
                        # 解析消息
                        state = parse_ur_realtime_data(message)
                        if state:
                            # 通知所有回调函数
                            for callback in self.callbacks:
                                try:
                                    callback(state)
                                except Exception as e:
                                    print(f"回调函数执行错误：{e}")
                    else:
                        # 数据不完整，等待更多数据
                        break
                        
            except socket.timeout:
                # 超时是正常的，继续循环
                continue
            except Exception as e:
                print(f"监控循环错误：{e}")
                break
        
        # 清理
        self.monitoring = False
        if self.socket:
            self.socket.close()
        print("状态监控已停止")
    
    def add_callback(self, callback):
        """添加状态回调函数"""
        self.callbacks.append(callback)
    
    def stop_monitoring(self):
        """停止监控"""
        self.monitoring = False

这个监控器的主要特点：

• 异步处理：在后台线程中处理数据接收，不阻塞主程序
• 回调机制：允许注册多个回调函数处理状态更新
• 错误恢复：连接断开时会自动停止，可以重新启动

5. 错误处理与调试技巧

即使按照最佳实践来写代码，在实际运行中还是会遇到各种问题。好的错误处理机制和调试技巧能大大缩短解决问题的时间。

5.1 常见错误类型与解决方案

根据我的经验，UR机器人Socket通信中最常见的问题可以分为以下几类：

连接类错误：

• 连接被拒绝：检查机器人IP和端口是否正确，确认机器人控制器已启动Socket服务
• 连接超时：检查网络连通性，确认防火墙设置
• 连接不稳定：检查网线质量，避免使用无线网络

指令执行错误：

• 指令无响应：检查脚本语法，确认发送的指令格式正确
• 执行结果不符合预期：检查关节位置或工具位姿是否可达
• 运动过程中停止：检查是否触发了安全限制或关节限位

数据接收错误：

• 数据解析错误：检查数据包格式是否与控制器版本匹配
• 数据丢失：检查网络带宽，减少不必要的数据传输
• 数据延迟：优化接收逻辑，避免在回调函数中执行耗时操作

5.2 实用的调试工具和技巧

日志记录系统：

import logging
import datetime

class URDebugLogger:
    def __init__(self, log_file="ur_communication.log"):
        # 配置日志
        logging.basicConfig(
            level=logging.DEBUG,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler(log_file),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    
    def log_command(self, command, success=True):
        """记录指令发送"""
        status = "成功" if success else "失败"
        self.logger.info(f"发送指令: {command[:50]}... [{status}]")
    
    def log_state(self, state):
        """记录状态更新"""
        # 选择性记录，避免日志过大
        if state.get('joint_positions'):
            positions = state['joint_positions']
            self.logger.debug(f"关节位置: {positions}")
    
    def log_error(self, error, context=""):
        """记录错误"""
        self.logger.error(f"{context}: {error}")

网络诊断工具：

def diagnose_network_connection(ip, port=30003):
    """诊断网络连接问题"""
    import subprocess
    import platform
    
    system = platform.system()
    
    print(f"诊断连接到 {ip}:{port}")
    
    # 1. 检查基本连通性
    print("\n1. 检查基本连通性...")
    if system == "Windows":
        result = subprocess.run(['ping', '-n', '4', ip], 
                               capture_output=True, text=True)
    else:
        result = subprocess.run(['ping', '-c', '4', ip], 
                               capture_output=True, text=True)
    
    if "请求超时" in result.stdout or "100% packet loss" in result.stdout:
        print("❌ 无法ping通机器人")
        return False
    else:
        print("✅ 可以ping通机器人")
    
    # 2. 检查端口是否开放
    print("\n2. 检查端口是否开放...")
    try:
        test_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        test_sock.settimeout(2)
        result = test_sock.connect_ex((ip, port))
        test_sock.close()
        
        if result == 0:
            print(f"✅ 端口 {port} 开放")
            return True
        else:
            print(f"❌ 端口 {port} 未开放")
            return False
    except Exception as e:
        print(f"❌ 端口检查失败: {e}")
        return False

交互式调试工具： 有时候，最简单的调试方法就是直接发送指令看看机器人的反应。我经常使用一个简单的交互式工具：

def interactive_ur_shell(ip="192.168.1.100", port=30003):
    """交互式UR指令发送工具"""
    sock = connect_to_ur(ip, port)
    if not sock:
        print("连接失败")
        return
    
    print(f"已连接到 {ip}:{port}")
    print("输入UR脚本指令（输入'quit'退出）")
    print("输入'help'查看帮助")
    
    while True:
        try:
            # 获取用户输入
            user_input = input("\nUR> ").strip()
            
            if user_input.lower() == 'quit':
                print("退出交互模式")
                break
            elif user_input.lower() == 'help':
                print("可用命令:")
                print("  quit - 退出")
                print("  help - 显示帮助")
                print("  movej [j1,j2,j3,j4,j5,j6] - 关节运动")
                print("  movel [x,y,z,rx,ry,rz] - 线性运动")
                print("  或直接输入UR脚本指令")
                continue
            
            # 发送指令
            if user_input.startswith("movej"):
                # 解析关节位置
                pass
            elif user_input.startswith("movel"):
                # 解析工具位姿
                pass
            
            # 添加end标记
            if not user_input.endswith("\n"):
                user_input += "\n"
            
            sock.send(user_input.encode('utf-8'))
            print(f"已发送: {user_input.strip()}")
            
        except KeyboardInterrupt:
            print("\n用户中断")
            break
        except Exception as e:
            print(f"错误: {e}")
    
    sock.close()

5.3 性能优化建议

当系统运行稳定后，下一步就是优化性能。以下是一些经过验证的优化建议：

减少通信延迟的技巧：

1. 使用二进制协议：如果可能，使用UR的二进制协议而不是文本协议
2. 批量发送指令：将多个运动指令合并为一个脚本发送
3. 减少状态查询频率：只在需要时查询状态，而不是持续高频查询
4. 本地缓存状态：对于不频繁变化的状态，可以在本地缓存

内存和资源管理：

class OptimizedURController:
    def __init__(self):
        self.command_buffer = []  # 指令缓冲区
        self.last_send_time = 0
        self.send_interval = 0.01  # 10ms发送间隔
        
    def queue_command(self, command):
        """将指令加入队列"""
        self.command_buffer.append(command)
        
    def process_buffer(self):
        """处理缓冲区中的指令"""
        current_time = time.time()
        
        # 检查是否达到发送间隔
        if current_time - self.last_send_time < self.send_interval:
            return
            
        if self.command_buffer:
            # 合并缓冲区中的指令
            combined_script = self._combine_commands(self.command_buffer)
            
            # 发送合并后的指令
            self._send_script(combined_script)
            
            # 清空缓冲区
            self.command_buffer.clear()
            self.last_send_time = current_time
    
    def _combine_commands(self, commands):
        """合并多个指令为一个脚本"""
        script_lines = ["def combined_move():"]
        
        for cmd in commands:
            # 移除可能的换行符
            clean_cmd = cmd.strip()
            if clean_cmd:
                script_lines.append(f"  {clean_cmd}")
        
        script_lines.append("end")
        return "\n".join(script_lines)

这种缓冲机制可以：

• 减少网络通信次数：多个指令一次发送
• 平滑发送频率：避免突发的大量小数据包
• 提高系统响应性：主线程不会被阻塞在发送操作上

6. 实际应用案例与进阶技巧

理论讲得再多，不如看几个实际的应用案例。这里我分享几个在实际项目中用到的模式，希望能给你一些启发。

6.1 案例一：自动化测试系统

在一个产品质量检测项目中，我们需要控制UR机器人抓取产品，移动到各个检测工位，最后根据检测结果进行分类。

系统架构：

产品上料 → 机器人抓取 → 视觉检测1 → 尺寸测量 → 视觉检测2 → 分类放置

通信模式设计：

class QualityInspectionSystem:
    def __init__(self, robot_ip):
        self.robot_ip = robot_ip
        self.command_socket = None  # 30003端口，发送指令
        self.state_socket = None    # 30002端口，接收状态
        self.current_state = {}
        
    def initialize(self):
        """初始化系统"""
        # 建立指令连接
        self.command_socket = connect_to_ur(self.robot_ip, 30003)
        
        # 建立状态监控连接
        self.state_socket = connect_to_ur(self.robot_ip, 30002)
        
        # 启动状态监控线程
        self._start_state_monitor()
        
        # 移动到初始位置
        self.move_to_home()
    
    def perform_inspection_cycle(self, product_id):
        """执行一个完整的检测周期"""
        try:
            # 1. 移动到抓取位置
            self.move_to_pick_position()
            
            # 2. 抓取产品
            self.gripper_activate(True)
            time.sleep(0.5)  # 等待抓取完成
            
            # 3. 移动到各个检测工位
            inspection_stations = [
                "vision_station_1",
                "measurement_station", 
                "vision_station_2"
            ]
            
            inspection_results = {}
            
            for station in inspection_stations:
                # 移动到检测位置
                self.move_to_station(station)
                
                # 执行检测（与外部系统通信）
                result = self.perform_inspection(station, product_id)
                inspection_results[station] = result
                
                # 如果检测失败，提前结束
                if not result["passed"]:
                    break
            
            # 4. 根据结果分类放置
            final_result = self.evaluate_results(inspection_results)
            self.place_product(final_result["category"])
            
            # 5. 释放产品
            self.gripper_activate(False)
            
            return {
                "product_id": product_id,
                "result": final_result,
                "timestamp": time.time()
            }
            
        except Exception as e:
            self._handle_error(e)
            return None
    
    def move_to_station(self, station_name):
        """移动到指定工位"""
        # 从配置中获取位置信息
        position = self.config["stations"][station_name]
        
        # 创建运动指令
        if position["type"] == "joint":
            command = create_movej_command(position["values"])
        else:
            command = create_movel_command(position["values"])
        
        # 发送指令并等待完成
        self.send_and_wait(command)

这个系统的关键设计点：

• 状态监控与指令发送分离：使用不同的Socket连接，避免互相干扰
• 错误处理完善：每个步骤都有错误检测和恢复机制
• 配置驱动：工位位置等信息从配置文件读取，便于调整

6.2 案例二：协同作业系统

在另一个项目中，我们需要两台UR机器人协同工作，一台负责上料，一台负责加工。

协同控制策略：

class CollaborativeRobotSystem:
    def __init__(self, robot1_ip, robot2_ip):
        self.robot1 = URController(robot1_ip)
        self.robot2 = URController(robot2_ip)
        self.coordination_lock = threading.Lock()
        self.shared_state = {
            "workpiece_ready": False,
            "processing_done": False,
            "current_step": 0
        }
    
    def coordinate_operation(self):
        """协调两台机器人的操作"""
        # 机器人1：上料
        def robot1_task():
            # 等待开始信号
            while not self.shared_state["current_step"] == 1:
                time.sleep(0.1)
            
            # 执行上料操作
            self.robot1.pick_from_feeder()
            self.robot1.move_to_handover_position()
            
            # 设置工件就绪标志
            with self.coordination_lock:
                self.shared_state["workpiece_ready"] = True
            
            # 等待机器人2取走工件
            while self.shared_state["workpiece_ready"]:
                time.sleep(0.1)
        
        # 机器人2：加工
        def robot2_task():
            # 等待工件就绪
            while not self.shared_state["workpiece_ready"]:
                time.sleep(0.1)
            
            # 取走工件
            self.robot2.take_from_robot1()
            
            # 清除工件就绪标志
            with self.coordination_lock:
                self.shared_state["workpiece_ready"] = False
            
            # 执行加工
            self.robot2.process_workpiece()
            
            # 设置加工完成标志
            with self.coordination_lock:
                self.shared_state["processing_done"] = True
        
        # 启动两个任务
        thread1 = threading.Thread(target=robot1_task)
        thread2 = threading.Thread(target=robot2_task)
        
        # 设置开始信号
        with self.coordination_lock:
            self.shared_state["current_step"] = 1
        
        thread1.start()
        thread2.start()
        
        thread1.join()
        thread2.join()

协同作业的关键挑战：

• 同步问题：确保两台机器人的动作协调
• 安全考虑：避免碰撞和干涉
• 错误恢复：一台机器人出错时，另一台能安全停止

6.3 高级技巧：动态轨迹生成

对于需要实时调整轨迹的应用，我们可以动态生成运动指令：

class DynamicTrajectoryGenerator:
    def __init__(self, robot_ip):
        self.robot_ip = robot_ip
        self.socket = None
        self.trajectory_points = []
        
    def generate_circular_trajectory(self, center, radius, points=50):
        """生成圆形轨迹"""
        trajectory = []
        
        for i in range(points):
            angle = 2 * math.pi * i / points
            
            # 计算位置
            x = center[0] + radius * math.cos(angle)
            y = center[1] + radius * math.sin(angle)
            z = center[2]
            
            # 保持姿态不变
            rx, ry, rz = center[3], center[4], center[5]
            
            trajectory.append([x, y, z, rx, ry, rz])
        
        return trajectory
    
    def execute_trajectory(self, trajectory, velocity=0.1, blend_radius=0.01):
        """执行轨迹"""
        if not self.socket:
            self.socket = connect_to_ur(self.robot_ip)
        
        # 生成脚本
        script_lines = ["def dynamic_trajectory():"]
        
        for i, point in enumerate(trajectory):
            x, y, z, rx, ry, rz = point
            
            # 对于最后一个点，不使用融合半径
            if i == len(trajectory) - 1:
                blend = 0
            else:
                blend = blend_radius
            
            line = f"  movel(p[{x:.3f}, {y:.3f}, {z:.3f}, {rx:.3f}, {ry:.3f}, {rz:.3f}], v={velocity}, r={blend})"
            script_lines.append(line)
        
        script_lines.append("end")
        
        # 发送脚本
        full_script = "\n".join(script_lines)
        self.socket.send(full_script.encode('utf-8'))

这种动态轨迹生成技术在以下场景特别有用：

• 视觉引导：根据相机检测结果实时调整运动路径
• 力控应用：根据力传感器反馈调整位置
• 复杂轨迹：需要数学计算生成的轨迹

7. 安全与可靠性考虑

在工业环境中，安全永远是第一位的。UR机器人的Socket通信虽然强大，但如果使用不当，也可能带来安全隐患。

7.1 安全通信实践

输入验证： 所有发送到机器人的指令都应该经过严格的验证：

def validate_ur_command(command):
    """验证UR指令的安全性"""
    # 检查是否为字符串
    if not isinstance(command, str):
        raise ValueError("指令必须是字符串")
    
    # 检查长度
    if len(command) > 10000:  # 防止过长的指令
        raise ValueError("指令过长")
    
    # 检查危险指令
    dangerous_keywords = [
        "halt",
        "kill",
        "shutdown",
        "force", 
        "override"
    ]
    
    command_lower = command.lower()
    for keyword in dangerous_keywords:
        if keyword in command_lower:
            raise SecurityError(f"指令包含危险关键词: {keyword}")
    
    # 检查语法（简化版）
    if not command.strip().endswith("\n"):
        command += "\n"
    
    return command

连接安全：

class SecureURConnection:
    def __init__(self, ip, port=30003, max_retries=3):
        self.ip = ip
        self.port = port
        self.max_retries = max_retries
        self.socket = None
        self.connection_time = None
        self.last_heartbeat = None
        
    def connect_with_retry(self):
        """带重试的连接"""
        for attempt in range(self.max_retries):
            try:
                self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                self.socket.settimeout(5.0)
                self.socket.connect((self.ip, self.port))
                
                self.connection_time = time.time()
                self.last_heartbeat = time.time()
                
                print(f"连接成功 (尝试 {attempt + 1} 次)")
                return True
                
            except Exception as e:
                print(f"连接尝试 {attempt + 1} 失败: {e}")
                if attempt < self.max_retries - 1:
                    time.sleep(2 ** attempt)  # 指数退避
                else:
                    print("达到最大重试次数，连接失败")
                    return False
    
    def send_with_ack(self, command, timeout=5.0):
        """发送指令并等待确认"""
        # 验证指令
        validated_command = validate_ur_command(command)
        
        # 发送指令
        self.socket.send(validated_command.encode('utf-8'))
        
        # 等待确认（这里需要根据实际协议实现）
        # UR机器人通常不会发送明确的ACK，但我们可以通过其他方式验证
        
        # 更新心跳时间
        self.last_heartbeat = time.time()
        
        return True
    
    def heartbeat_check(self):
        """心跳检查"""
        current_time = time.time()
        
        # 如果超过30秒没有通信，发送心跳
        if current_time - self.last_heartbeat > 30:
            try:
                # 发送一个无害的指令保持连接
                self.send_with_ack("get_actual_tcp_pose()\n")
                return True
            except Exception as e:
                print(f"心跳失败: {e}")
                return False
        
        return True

7.2 异常处理策略

完善的异常处理策略应该包括：

分级错误处理：

class URExceptionHandler:
    ERROR_LEVELS = {
        "INFO": 1,
        "WARNING": 2, 
        "ERROR": 3,
        "CRITICAL": 4
    }
    
    def handle_exception(self, exception, context=""):
        """处理异常"""
        error_level = self._classify_exception(exception)
        
        # 根据错误级别采取不同措施
        if error_level == self.ERROR_LEVELS["INFO"]:
            # 记录日志，继续运行
            self.log_info(exception, context)
            
        elif error_level == self.ERROR_LEVELS["WARNING"]:
            # 记录警告，尝试恢复
            self.log_warning(exception, context)
            self._try_recover()
            
        elif error_level == self.ERROR_LEVELS["ERROR"]:
            # 记录错误，停止当前操作
            self.log_error(exception, context)
            self._stop_current_operation()
            
        elif error_level == self.ERROR_LEVELS["CRITICAL"]:
            # 记录严重错误，紧急停止
            self.log_critical(exception, context)
            self._emergency_stop()
    
    def _classify_exception(self, exception):
        """分类异常"""
        if isinstance(exception, ConnectionError):
            return self.ERROR_LEVELS["ERROR"]
        elif isinstance(exception, TimeoutError):
            return self.ERROR_LEVELS["WARNING"]
        elif isinstance(exception, ValueError):
            return self.ERROR_LEVELS["WARNING"]
        elif isinstance(exception, SecurityError):
            return self.ERROR_LEVELS["CRITICAL"]
        else:
            return self.ERROR_LEVELS["ERROR"]

自动恢复机制：

class AutoRecoveryController:
    def __init__(self, robot_ip):
        self.robot_ip = robot_ip
        self.connection = None
        self.recovery_attempts = 0
        self.max_recovery_attempts = 5
        
    def ensure_connection(self):
        """确保连接正常"""
        if not self.connection or not self._check_connection_alive():
            print("连接异常，尝试恢复...")
            self._recover_connection()
    
    def _check_connection_alive(self):
        """检查连接是否存活"""
        try:
            # 尝试发送一个简单的指令
            self.connection.send("get_actual_tcp_pose()\n".encode('utf-8'))
            
            # 尝试接收响应（如果有）
            self.connection.settimeout(1.0)
            try:
                data = self.connection.recv(1024)
                return True
            except socket.timeout:
                # 没有响应不一定代表连接断开
                return True
                
        except Exception:
            return False
    
    def _recover_connection(self):
        """恢复连接"""
        self.recovery_attempts += 1
        
        if self.recovery_attempts > self.max_recovery_attempts:
            print("达到最大恢复尝试次数，需要人工干预")
            self._notify_operator()
            return False
        
        try:
            # 关闭旧连接
            if self.connection:
                self.connection.close()
            
            # 建立新连接
            self.connection = connect_to_ur(self.robot_ip)
            
            if self.connection:
                print("连接恢复成功")
                self.recovery_attempts = 0
                return True
            else:
                print("连接恢复失败")
                return False
                
        except Exception as e:
            print(f"恢复连接时出错: {e}")
            return False

在实际项目中，我发现最有效的错误处理策略是“防御性编程”——假设任何操作都可能失败，并为每种失败情况准备恢复方案。比如，每次发送指令前都检查连接状态，重要操作都有超时机制，所有异常都被捕获并妥善处理。

Socket通信的稳定性不仅取决于代码质量，还受到网络环境、机器人状态、甚至车间电磁干扰的影响。有一次我们在调试时发现，每天下午3点左右通信就会不稳定，最后发现是隔壁车间的设备定时启动造成了电网波动。这种问题通过代码层面的优化很难完全解决，但可以通过增加重试机制、心跳检测、状态监控等方式提高系统的鲁棒性。

对于关键的生产应用，我建议至少实现以下安全措施：

1. 双重验证：重要指令执行前，通过不同通道验证机器人状态
2. 超时保护：所有操作都有超时限制，防止程序挂起
3. 状态同步：定期同步控制端和机器人的状态信息
4. 手动干预接口：在紧急情况下可以快速切换到手动控制
5. 详细日志：记录所有操作和异常，便于问题追溯

这些经验都是在实际项目中一点点积累起来的。刚开始可能觉得繁琐，但当系统稳定运行几个月不出问题时，你就会发现这些投入是值得的。毕竟在工业自动化领域，稳定性往往比功能丰富更重要。