Java / Python中的快速IPC / Socket通信 [英] Fast IPC/Socket communication in Java/Python

问题描述

两个进程（Java和Python）需要在我的应用程序中进行通信。我注意到套接字通信占用了93％的运行时间。为什么沟通这么慢？我应该寻找套接字通信的替代方案，还是可以更快地做到这一点？

Two processes (Java and Python) need to communicate in my application. I noticed that the socket communication takes 93% of the run time. Why is communication so slow? Should I be looking for alternatives to socket communication or can this be made faster?

更新：我发现了一个简单的修复方法。似乎缓冲输出流由于某种未知原因而没有真正缓冲。所以，我现在将所有数据放入客户端/服务器进程的字符串缓冲区中。我在flush方法中将它写入套接字。

Update: I discovered a simple fix. It seems like the Buffered output stream is not really buffered for some unknown reason. So, I now put all data into string buffers in both client/server processes. I write it to the socket in the flush method.

我仍然对使用共享内存在进程之间快速交换数据的示例感兴趣。

I'm still interested in an example of the usage of shared memory to exchange data quickly between processes.

一些额外信息：

1. 申请中的邮件大小大多数时间不到64kb 。


2. 服务器是Java，客户端是用Python编写的。


3. 套接字IPC实现如下：它需要50个周期发送200个字节！这必须太高了。如果我在5000个周期内发送2个字节，则需要花费更少的时间。


4. 两个进程都在一台Linux机器上运行。


5. 在实际应用程序中每个周期都会对客户端的iFid.write（）进行大约10次调用。


6. 这是在Linux系统上完成的。

1. Message size in the applicaiton is under 64kb most of the time.
2. The server is in Java, the client is written in Python.
3. Socket IPC is implemented below: it takes 50 cycles sending 200 bytes ! This has got to be too high. If I send 2 bytes in 5000 cycles, it takes a lot less time.
4. Both processes run on one Linux machine.
5. In the real application about 10 calls to client's iFid.write() are made each cycle.
6. This is done on a Linux system.

这是服务器端：

public class FastIPC{
    public PrintWriter out;
    BufferedReader in;
    Socket socket = null;
    ServerSocket serverSocket = null;


    public FastIPC(int port) throws Exception{
        serverSocket = new ServerSocket(port);
        socket = serverSocket.accept();
        out = new PrintWriter(new BufferedWriter(new OutputStreamWriter(socket.getOutputStream())), true);
        in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
    }

    public void send(String msg){
        out.println(msg); // send price update to socket
    }

    public void flush(){
        out.flush();
    }

    public String recv() throws Exception{
        return in.readLine();
    }

    public static void main(String[] args){
        int port = 32000;
        try{
            FastIPC fip = new FastIPC(port);
            long start = new Date().getTime();
            System.out.println("Connected.");
            for (int i=0; i<50; i++){
                for(int j=0; j<100; j++)
                    fip.send("+");
                fip.send(".");
                fip.flush();
                String msg = fip.recv();
            }
            long stop = new Date().getTime();
            System.out.println((double)(stop - start)/1000.);
        }catch(Exception e){
            System.exit(1);
        }
    }
}

客户端是：

import sys
import socket

class IPC(object):
    def __init__(self):
        self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.s.connect(("localhost", 32000))
        self.fid = self.s.makefile() # file wrapper to read lines
        self.listenLoop() # wait listening for updates from server

    def listenLoop(self):
        fid = self.fid
        print "connected"
        while True:
            while True:
                line = fid.readline()
                if line[0]=='.':
                    break
            fid.write('.\n')
            fid.flush()

if __name__ == '__main__':
    st = IPC()

推荐答案

你有一些选项。由于您使用的是Linux，因此可以使用UNIX域套接字。或者，您可以将数据序列化为ASCII或JSon或其他格式，并通过管道，SHM（共享内存段），消息队列，DBUS或类似方式提供数据。由于这些IPC机制具有不同的性能特征，因此值得考虑您拥有的数据类型。有一个起草USENIX论文，并对各种权衡进行了很好的分析。值得一读。

You have a number of options. Since you are using Linux you could use UNIX domain sockets. Or, you could serialise the data as ASCII or JSon or some other format and feed it through a pipe, SHM (shared memory segment), message queue, DBUS or similar. It's worth thinking about what sort of data you have, as these IPC mechanisms have different performance characteristics. There's a draft USENIX paper with a good analysis of the various trade-offs that is worth reading.

既然你说（在这个答案的评论中）你更喜欢使用SHM，这里有一些代码示例可以让你开始。使用Python posix_ipc 库：

Since you say (in the comments to this answer) that you prefer to use SHM, here are some code samples to start you off. Using the Python posix_ipc library:

import posix_ipc # POSIX-specific IPC
import mmap      # From Python stdlib

class SharedMemory(object):
    """Python interface to shared memory. 
    The create argument tells the object to create a new SHM object,
    rather than attaching to an existing one.
    """

    def __init__(self, name, size=posix_ipc.PAGE_SIZE, create=True):
        self.name = name
        self.size = size
        if create:
            memory = posix_ipc.SharedMemory(self.name, posix_ipc.O_CREX,
                                            size=self.size)
        else:
            memory = posix_ipc.SharedMemory(self.name)
        self.mapfile = mmap.mmap(memory.fd, memory.size)
        os.close(memory.fd)
        return

    def put(self, item):
        """Put item in shared memory.
        """
        # TODO: Deal with the case where len(item) > size(self.mapfile)
        # TODO: Guard this method with a named semaphore
        self.mapfile.seek(0)
        pickle.dump(item, self.mapfile, protocol=2)
        return

    def get(self):
        """Get a Python object from shared memory.
        """
        # TODO: Deal with the case where len(item) > size(self.mapfile)
        # TODO: Guard this method with a named semaphore
        self.mapfile.seek(0)
        return pickle.load(self.mapfile)

    def __del__(self):
        try:
            self.mapfile.close()
            memory = posix_ipc.SharedMemory(self.name)
            memory.unlink()
        except:
            pass
        return    

对于Java方面，您想要创建相同的类，尽管我在评论中说过 JTux 似乎提供了相同的功能，您需要的API位于 UPosixIPC 类。

For the Java side you want to create the same class, despite what I said in the comments JTux seems to provide the equivalent functionality and the API you need is in UPosixIPC class.

下面的代码概述了你需要实现的东西。但是，有几个缺失 - 异常处理很明显，als o一些标志（在 UConstant 中找到它们），你会想要添加一个信号量保护 put / 获取方法。但是，这应该让你走上正轨。请记住， mmap 或内存映射文件是一段RAM的文件类接口。因此，您可以将其文件描述符用作普通文件的 fd 。

The code below is an outline of the sort of thing you need to implement. However, there are several things missing -- exception handling is the obvious one, also some flags (find them in UConstant), and you'll want to add in a semaphore to guard the put / get methods. However, this should set you on the right track. Remember that an mmap or memory-mapped file is a file-like interface to a segment of RAM. So, you can use its file descriptor as if it were the fd of a normal file.

import jtux.*;

class SHM {

    private String name;
    private int size;
    private long semaphore;
    private long mapfile; // File descriptor for mmap file

    /* Lookup flags and perms in your system docs */
    public SHM(String name, int size, boolean create, int flags, int perms) {
        this.name = name;
        this.size = size;
        int shm;
        if (create) {
            flags = flags | UConstant.O_CREAT;
            shm = UPosixIPC.shm_open(name, flags, UConstant.O_RDWR);
        } else {
            shm = UPosixIPC.shm_open(name, flags, UConstant.O_RDWR);
        }
        this.mapfile = UPosixIPC.mmap(..., this.size, ..., flags, shm, 0);
        return;
    }


    public void put(String item) {
        UFile.lseek(this.mapfile(this.mapfile, 0, 0));
        UFile.write(item.getBytes(), this.mapfile);
        return;
    }


    public String get() {    
        UFile.lseek(this.mapfile(this.mapfile, 0, 0));
        byte[] buffer = new byte[this.size];
        UFile.read(this.mapfile, buffer, buffer.length);
        return new String(buffer);
    }


    public void finalize() {
        UPosix.shm_unlink(this.name);
        UPosix.munmap(this.mapfile, this.size);
    }

}

这篇关于Java / Python中的快速IPC / Socket通信的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！