如何在两个不同的线程之间切换？

Question

我正在与ROS中的串行通信合作。到目前为止，我一直在使用C ++，但需要将代码作为Python中的其他模块API将代码移植到python。我已经成功创建了一个发布者和订户NPDE，但是当我使用序列时，我可以同时写或阅读？因此，我决定为我的系统制作两个不同的线程，其中一个用于读取数据，一个用于发送数据。一切正常，但是当数据再次进来时，第二个线程没有结束，并且在无限的循环中，因此我可以接收数据，但是来自主题的传入数据并未通过调制解调器发送带有写入线程的调制解调器。

有人可以看看这个问题，以了解为什么程序粘在接收线程上吗？ PS：您可能无法运行代码，因为需要将实际调制解调器连接到系统。

#subscriber_node.py
 
import threading
import sys
import time

from time import sleep
from queue import Queue, Empty

# The main ROS Python client library
import rclpy # Import the ROS client library for Python

# The main node class which will be used to create custom nodes
from rclpy.node import Node # Enables the use of rclpy's Node class

# This is for standard message types when defining publishers
from std_msgs.msg import String 

# This is whoi python APIs.
from py_whoi.micromodem import *
from py_whoi.messageparams import *
from py_whoi.messageparser import *
from py_whoi.unifiedlog import UnifiedLog

# Open a connection to modem
# unilog = UnifiedLog(console_log_level='INFO')

modem_a = Micromodem(name="MM2")#, unified_log=unilog )
modem_serial_port = "/dev/ttyUSB0"  
modem_baudrate = 19200
modem_source_id = 0
modem_dest_id = 1
modem_a.connect(modem_serial_port, modem_baudrate)

mutex = threading.Lock()

class WHOISubscriber(Node):
    def __init__(self):
        super().__init__("whoi_sub_node")

        # Create subscriber(s)    
     
        # The node subscribes to messages of type std_msgs/Float64MultiArray, over a topic named:
        #   /object_msg
        # The callback function is called as soon as a message is received.
        # The maximum number of queued messages is 10.
        self.subscription_1 = self.create_subscription(
          String,
          '/object_msg',
          self.listener_callback,
          10)
        self.subscription_1  # prevent unused variable warning

        # Create publisher(s)  
         
        # This node publishes the position in robot frame coordinates.
        # The node publishes to messages of type std_msgs/Float64MultiArray, over a topic named:
        #   /pos_in_robot_base_frame
        # Maximum queue size of 10. 
        self.publisher_pos_robot_frame = self.create_publisher(String, '/pos_in_robot_base_frame', 10)
        timer_period = 3.0  # seconds
        self.timer = self.create_timer(timer_period, self.publish_position)
        self.i = 0

       
        @property
        def is_connected(self):
            return self._serialport.isOpen()
       
         
    def publish_position(self):
        """
        Callback function.
        Publish Hello World message to the topic /pos_in_robot_base_frame
        for every 3 seconds. Change the message content accordingly.
        """
        # msg = Float64MultiArray() # Create a message of this type 
        msg = String() # Create a message of this type 
        msg.data = 'Hello World: %d from subscriber whoi.' % self.i # Store the object's position
        self.publisher_pos_robot_frame.publish(msg) # Publish the position to the topic 
        self.get_logger().info('Publishing publish_position: "%s"' % msg.data)
        self.i += 1

    def send_data(self, msg):
        """
        sender function.
        Function to send the data from the whoi modem using APIs. It takes the msg data coming from the topic
        /object_msg, prepare and send it to the modem. It uses the command "send" to send the data. i.e It uses
        the legacy packet method in which you need to tell the modem first the cycleinit and then send the data 
        which is in the hex format, which would send acoustically to the destination node  in the system.
        This serves as UDP packet. Sleep timer has been used to wait the time taken to send the data and actually decode
        received data.
        """
        try:
            # mutex.acquire(blocking=True)
            data_to_be_send = bytes(msg, 'utf-8')
            #modem_a.send_cycleinit((cycleinfo(src_id=0, dest_id=1, rate_num=0, ack=1, num_frames=1))
            modem_a.send_cycleinit(cycleinfo = CycleInfo(modem_source_id, modem_dest_id, 0, 1, 1)) # (src_id, dest_id, rate_num, ack, 1)
            sleep(1)
            # modem_a.send_frame(dataframe = DataFrame(src=0, dest=1, rate_num=0, ack=1, data=data_to_be_send))
            modem_a.send_frame(dataframe = DataFrame(modem_source_id, modem_dest_id, 0, 1, data_to_be_send))  
            sleep(1)
            self.get_logger().info('WHOI Modem sent data acoustically.')
            # mutex.release()
            print(data_to_be_send)
            
        except:                       
            modem_a.disconnect()
            self.get_logger().info('WHOI Modem disconnected after sending data in send data function.')
    

    def rec_data(self):
        """
        receiver function.
        Function to receive the data from the modem using APIs. It takes the msg data coming on the modem 
        acoustically in the newly created queue. It basically waits for the CARXD message from the modem. 
        Once it receives the CARXD message, it decodes the data and prints the data.
        i.e It uses the acomms micromodem class. 
        """
        # modem_a = Micromodem(name="MM2")
        # modem_a.connect(modem_serial_port, modem_baudrate)

        try:
            """"This will create a new queue for Rx messages on micromodem and attach a new queue with
            existing attach_incoming_msg_queue function and print the new incoming messages from new queue."
            """
            # mutex.acquire(blocking=True)
            modem_rx_queue = Queue()
            modem_a.attach_incoming_msg_queue(modem_rx_queue)
            while True:
                try:
                    """This will get the message from the queue and grab the first message and split the message in parts.
                    """
                    # if self._serialport.isOpen():
                    msg = modem_rx_queue.get(block=True, timeout=0.1)
                    
                    msgParts = str(msg).split(',')  # Need to convert the message from list to string
                    msg_type = msgParts[0]
                    msg_params = msgParts[5:]
                    # ['params'] = msgParts[1:]
                    # ['raw'] = raw

                    # to check if the message is a valid message or not empty message
                    if msg_type=="$CARXD" and msg_params != None and len(msg_params) > 0:
                        # print(msg_params) # print the message parameters that contained the actual message in Hex
                        Incoming_Rx_data = (msg_params[0]).split('*')[0]
                        print(Incoming_Rx_data) # Actual message in Hex
                        # mutex.release()
                        # self.get_logger().info('WHOI Modem Received Data acoustically: %s' % Incoming_Rx_data)
                        self.get_logger().info('WHOI Modem Received Data acoustically:')
                        # print(type(Incoming_Rx_data)) 
                    else: # not connected
                        sleep(0.2) # Wait half a second, try again.                   
                        
                except Empty:
                    pass
        except KeyboardInterrupt:
            print("\n\nExiting")
            modem_a.disconnect()
            self.get_logger().info('WHOI Modem disconnected after receiving.')
            sys.exit(0)

    def listener_callback(self, msg):
        self.get_logger().info('I got: "%s" on /object_msg' % msg.data)
        self.get_logger().info("Received a data from /object_msg and preparing for sending same data acoustically.")
        

        # Finally creating thread
        t1 = threading.Thread(target=self.rec_data)
        t2 = threading.Thread(target=self.send_data(str(msg.data)))

        # starting thread 1 - Comment this print statements if needed 
       
        t1.start()
        
        # t1.join()
        
        
        self.get_logger().info('thread 1 - receiving thread started.')

        # starting thread 2
        
        t2.start()
        
        self.get_logger().info('thread 2 - receiving thread started.')

        # wait until thread 1 is completely executed
        
        t1.join()
        
        
        self.get_logger().info('thread 1 - receiving thread joined.')
        
       
        t2.join()
       
        
        self.get_logger().info('thread 2 - sending thread joined.')


def main(args=None):
    # Initialise ROS client library
    # It can take commandline arguments or a context name
    # as input parameters, which we will not use currently.
    rclpy.init(args=args)

    my_sub = WHOISubscriber()

    print("[WHOI SUBSCRIBER] Waiting for data to be published over topic")

    try:
        # The `spin` function will keep the function from exiting (I assume
        # because it's all asyncronous now), until a KeyboardInterrupt.
        rclpy.spin(my_sub)

    except KeyboardInterrupt:
        # Kill the node
        my_sub.destroy_node()

        # Shutdown and disconnect the client library
        rclpy.shutdown()


if __name__ == "__main__":
    main()

输出

[WHOI SUBSCRIBER] Waiting for data to be published over topic
[INFO] [1652791325.779022111] [whoi_sub_node]: Publishing publish_position: "Hello World: 0 from subscriber whoi."
[INFO] [1652791328.757283538] [whoi_sub_node]: Publishing publish_position: "Hello World: 1 from subscriber whoi."
[INFO] [1652791331.757229112] [whoi_sub_node]: Publishing publish_position: "Hello World: 2 from subscriber whoi."
[INFO] [1652791334.756775223] [whoi_sub_node]: Publishing publish_position: "Hello World: 3 from subscriber whoi."
[INFO] [1652791337.757394185] [whoi_sub_node]: Publishing publish_position: "Hello World: 4 from subscriber whoi."
[INFO] [1652791340.097852311] [whoi_sub_node]: I got: "WNrnTtogMFN04v2KOYJLiMzkCh6R" on /object_msg
[INFO] [1652791340.098774010] [whoi_sub_node]: Received a data from /object_msg and preparing for sending same data acoustically.
[INFO] [1652791344.104244151] [whoi_sub_node]: WHOI Modem sent data acoustically.
b'WNrnTtogMFN04v2KOYJLiMzkCh6R'
000102030405060708090A0B0C0D0E0FA55AFFEEDDCCBBAA90807060504055AF
[INFO] [1652791365.446508085] [whoi_sub_node]: WHOI Modem Received Data acoustically:
000102030405060708090A0B0C0D0E0FA55AFFEEDDCCBBAA90807060504055AF
[INFO] [1652791367.446508087] [whoi_sub_node]: WHOI Modem Received Data acoustically:
[INFO] [1652791381.305565989] [whoi_sub_node]: thread 1 - receiving thread started.
[INFO] [1652791381.306729773] [whoi_sub_node]: thread 2 - receiving thread started.
[INFO] [1652791381.307589789] [whoi_sub_node]: thread 1 - receiving thread joined.
[INFO] [1652791381.308452921] [whoi_sub_node]: thread 2 - sending thread joined.
[INFO] [1652791381.309906606] [whoi_sub_node]: Publishing publish_position: "Hello World: 5 from subscriber whoi."
[INFO] [1652791381.311024559] [whoi_sub_node]: I got: "tewPB8BW0x8uNScphIzfu9Mpokhc" on /object_msg
[INFO] [1652791381.311877118] [whoi_sub_node]: Received a data from /object_msg and preparing for sending same data acoustically.
[INFO] [1652791385.317304354] [whoi_sub_node]: WHOI Modem sent data acoustically.
b'tewPB8BW0x8uNScphIzfu9Mpokhc'
000102030405060708090A0B0C0D0E0FA55AFFEEDDCCBBAA90807060504055AF
[INFO] [1652791404.073748634] [whoi_sub_node]: WHOI Modem Received Data acoustically:
000102030405060708090A0B0C0D0E0FA55AFFEEDDCCBBAA90807060504055AF
[INFO] [1652791418.337479476] [whoi_sub_node]: thread 1 - receiving thread started.
[INFO] [1652791418.338659823] [whoi_sub_node]: thread 2 - receiving thread started.
[INFO] [1652791418.339525429] [whoi_sub_node]: thread 1 - receiving thread joined.
[INFO] [1652791418.340388951] [whoi_sub_node]: thread 2 - sending thread joined.
[INFO] [1652791418.341853726] [whoi_sub_node]: Publishing publish_position: "Hello World: 6 from subscriber whoi."
[INFO] [1652791418.342981213] [whoi_sub_node]: I got: "ofUMyED0BbPfueF64VsHy3vFCrKd" on /object_msg
[INFO] [1652791418.343844862] [whoi_sub_node]: Received a data from /object_msg and preparing for sending same data acoustically.
[INFO] [1652791422.349254465] [whoi_sub_node]: WHOI Modem sent data acoustically.
b'ofUMyED0BbPfueF64VsHy3vFCrKd'
000102030405060708090A0B0C0D0E0FA55AFFEEDDCCBBAA90807060504055AF
[INFO] [1652791464.332005914] [whoi_sub_node]: WHOI Modem Received Data acoustically:

Answer 1

（str（msg.data）））在这里

# Finally creating thread
t1 = threading.Thread(target=self.rec_data)
t2 = threading.Thread(target=self.send_data(str(msg.data)))

我认为，您不应该通过self.send_data

t2 = threading.Thread(target=self.send_data, args=(str(msg.data),))

Answer 2

我认为，您可以将计时器与MultineReadedExecutor一起使用，而不是使用单独的线程。也许您可以控制使用Bool发送/接收数据何时。