杰瑞科技汇

Python DDE server如何实现与客户端通信?

Of course! Here is a comprehensive guide on how to create a Python DDE (Dynamic Data Exchange) server, including a full working example, explanations, and important considerations.

What is DDE?

DDE is an older inter-process communication (IPC) protocol from the early days of Windows (Windows 2.1, 1992). It allows applications to share data and execute commands. While largely superseded by more modern and robust technologies like OLE, COM, and .NET Remoting, it's still used in some legacy systems, particularly in scientific and industrial software (e.g., linking Python with LabVIEW, MATLAB, or old financial terminals).

Key DDE Concepts:

  • Server: The application that provides data or services.
  • Client: The application that requests data or services from the server.
  • Service: A name that the server advertises to identify itself (e.g., "MyPythonApp").
  • Topic: A category of data within a service (e.g., "LiveData", "Control").
  • Item: A specific piece of data within a topic (e.g., "Temperature", "StartProcess").

The Challenge: Native Python DDE Libraries

Python does not have a built-in library for DDE. You need to use a third-party library. The most common and well-maintained one is pywin32.

Prerequisites:

  1. Windows Operating System: DDE is a Windows-specific technology. This will not work on macOS or Linux.
  2. Python: A standard Windows installation of Python (3.6+ is recommended).
  3. pywin32: This library provides Python access to many Windows APIs, including DDE.

Installation:

Open your command prompt or terminal and install pywin32 using pip:

pip install pywin32

Creating a Python DDE Server

We will use the win32dde module from pywin32. The core idea is to:

  1. Create a DDEServer instance.
  2. Define a callback function that handles client requests.
  3. Start the server, which will begin listening for connections.

Full Example: A Simple DDE Server

This server will:

  • Advertise a service named PythonDDEServer.
  • Have one topic named Status.
  • Under the Status topic, it will provide two items:
    • Message: A static greeting.
    • Time: The current system time, which will be updated whenever the client requests it.
import win32dde
import win32serviceutil
import win32service
import win32event
import threading
import time
import datetime
# --- DDE Server Logic ---
class DDEServerApp:
    """
    A simple DDE server that provides status information.
    """
    def __init__(self):
        # The name of our DDE service. Clients will connect to this.
        self.service_name = "PythonDDEServer"
        # The topic we are offering data on.
        self.topic_name = "Status"
        # Create the DDE server instance
        # The callback function will be triggered on DDE events
        self.server = win32dde.DDEServer(self.service_name, self.on_dde_request)
    def on_dde_request(self, request):
        """
        This is the core callback function that handles all DDE requests.
        The 'request' object contains information about the client's request.
        We need to check the topic and item to provide the correct response.
        """
        topic = request.topic
        item = request.item
        print(f"DDE Request - Topic: '{topic}', Item: '{item}'")
        # We only handle requests for our specific topic
        if topic == self.topic_name:
            if item == "Message":
                # Return a simple string for the 'Message' item
                response = "Hello from Python DDE Server!"
                print(f"  -> Responding with: '{response}'")
                return response
            elif item == "Time":
                # Return the current time for the 'Time' item
                # DDE data is often passed as strings
                response = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                print(f"  -> Responding with: '{response}'")
                return response
        # If the topic or item is not recognized, return an empty string
        # or raise an exception to signal an error to the client.
        print(f"  -> Unknown Topic/Item. Responding with empty string.")
        return ""
    def start(self):
        """Starts the DDE server."""
        print(f"Starting DDE Server with service name: '{self.service_name}'")
        # The server runs in a separate thread
        self.server.start()
        print("Server is running. Press Ctrl+C to stop.")
    def stop(self):
        """Stops the DDE server."""
        print("Stopping DDE Server...")
        self.server.stop()
        print("Server stopped.")
# --- Main Execution ---
if __name__ == "__main__":
    try:
        dde_server = DDEServerApp()
        dde_server.start()
        # Keep the main thread alive while the server runs
        # In a real application, this might be a GUI event loop
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        dde_server.stop()

How to Run the Server

  1. Save the code above as a Python file (e.g., dde_server.py).
  2. Open a command prompt.
  3. Navigate to the directory where you saved the file.
  4. Run the script: python dde_server.py

You will see the output:

Starting DDE Server with service name: 'PythonDDEServer'
Server is running. Press Ctrl+C to stop.

The server is now active and waiting for clients.


Testing the DDE Server

You need a DDE client to test this. The easiest way is to use a built-in Windows tool: DDEML Test Client.

  1. Open the Test Client:

    • Press Win + R to open the Run dialog.
    • Type cmd and press Enter to open Command Prompt.
    • Run the following command:
      start ddeml
    • A window titled "DDEML Test Client" will appear.
  2. Connect to the Server:

    • In the DDEML Test Client window, click Connect.
    • Service: Enter PythonDDEServer
    • Topic: Enter Status
    • Click Connect. If successful, the connection status will show "Connected".
  3. Request Data (Advise Loop):

    • Advise Mode: This is the most powerful DDE feature. The client automatically receives updates whenever the server's data changes.
    • In the "Item" box, type Time.
    • Click Advise.
    • You will see the current time appear in the "Data" box, and it will automatically update every few seconds as the server provides new values.
    • Now, type Message into the "Item" box and click Advise. The static message "Hello from Python DDE Server!" will appear.
  4. Request Data (Data Mode):

    • Click Unadvise All to stop the automatic updates.
    • Type Time into the "Item" box and click Request. The current time will be fetched and displayed once.
    • Type Message and click Request to get the string.

Important Considerations and Advanced Topics

The Advise Loop (Automatic Updates)

In our example, the Time item works perfectly with an advise loop because the server always generates a new value when requested. The client doesn't need to know how the data is generated; it just gets the new value.

For an advise loop to be truly useful for live-updating data (e.g., sensor readings), your server should be able to proactively "poke" the client when data changes. This is more complex and often requires a timer in the server that periodically checks for new data and then triggers the callback to provide it.

Security

DDE is an old protocol with no built-in security. Any application on the same machine can connect to your server if it knows the service and topic names. Be very careful about running DDE servers on production machines or networks where untrusted applications might be present.

Running as a Windows Service

For a background server that needs to run continuously, you should wrap your DDE server in a Windows Service. The pywin32 library provides the win32serviceutil module to help with this. You would create a new class that inherits from win32service.Service, and in the SvcDoRun method, you would start your DDEServer instance.

Alternatives to DDE

For any new development, strongly consider using a more modern IPC mechanism:

  • HTTP/REST API: The most common and flexible approach. Your Python server can be a simple web server (e.g., using Flask or FastAPI), and clients can make HTTP requests to get data. This works across machines and is easy to debug with a web browser.
  • WebSockets: Excellent for real-time, two-way communication where the server needs to push data to the client (like a true advise loop).
  • gRPC / RPC: A high-performance, framework-agnostic way to call remote functions.
  • Named Pipes or Sockets: Lower-level but very fast IPC mechanisms, typically used within a single machine.

Summary

Feature Description
Library pywin32 (pip install pywin32)
Key Class win32dde.DDEServer
Key Method DDEServer(service_name, callback_function)
Callback Logic The callback function receives a request object with topic and item attributes. You return the data as a string.
Testing Tool ddeml (built-in Windows DDE test client).
Pros Simple for legacy Windows applications.
Cons Windows-only, insecure, outdated, limited data types (mostly strings).
分享:
扫描分享到社交APP
上一篇
下一篇