使用 POSIX 消息队列运行 JNA 示例

Question

我需要通过 POSIX 消息队列与 Java 应用程序和 C 进程通信，我想使用 JNA 来完成。

经过一些研究、阅读和您的帮助，我从一个尝试创建消息队列的简单 Java 应用程序开始。

/** Simple example of JNA interface mapping and usage. */
public class HelloJNAWorld {

    // This is the standard, stable way of mapping, which supports extensive
    // customization and mapping of Java to native types.

     public interface IPCLibrary extends Library {
        IPCLibrary INSTANCE = (IPCLibrary)
                Native.loadLibrary("c",IPCLibrary.class);


        int msgget(NativeLong key, int msgflg);

    }

    public static void main(String[] args) {

        int id = IPCLibrary.INSTANCE.msgget(new NativeLong(12500), 0600|1);


        if(id<0){
                System.out.println("Error creating message queue. Id:"+id);
                System.out.println(Native.getLastError());
        }else{
                System.out.println("Message queue id:" + idCola);
        }

    }
}

我认为msgctl 是最简单的映射方法，因为它只是int msgget(key_t key, int msgflag);。我假设我可以将key_t 映射为NativeLong，但msget 返回-1。所以我检查了lastError，返回的值为2，这意味着“没有这样的文件或 目录”根据errno代码。

你能帮我解决这个问题吗？也许key_t 应该以另一种方式映射？也许我需要更多的库或类似的东西？

Answer 1

由于没有人回答这个问题，而且有些人可能需要我当时需要的帮助，所以我在这里发布了我的测试类代码。 :-)

import com.sun.jna.Library;
import com.sun.jna.Native;
import com.sun.jna.NativeLong;
import com.sun.jna.Platform;
import com.sun.jna.Structure;

/** Simple example of JNA interface mapping and usage. */
public class HelloJNAWorld {

    // This is the standard, stable way of mapping, which supports extensive
    // customization and mapping of Java to native types.

    public interface CLibrary extends Library {
        CLibrary INSTANCE = (CLibrary)
        Native.loadLibrary((Platform.isWindows() ? "msvcrt" : "c"),CLibrary.class);


        void printf(String format, Object... args);


    }

    public interface IPCLibrary extends Library {
        IPCLibrary INSTANCE = (IPCLibrary)           
        Native.loadLibrary("c",IPCLibrary.class);



        class WaitQueue extends Structure{

        }


        // mapping msqid_ds structure
        class MsqidDs extends Structure{
            long msg_stime;       /* last msgsnd time */
            long msg_rtime;       /* last msgrcv time */
            long msg_ctime;       /* last change time */
            short msg_cbytes;
            short msg_qnum;
            short msg_qbytes;      /* max number of bytes on queue */
            short msg_lspid;       /* pid of last msgsnd */
            short msg_lrpid;       /* last receive pid */
        }

        // END mapping msqid_ds structure

        class MsgBuf extends Structure{
            NativeLong mtype; /* type of message */
            byte mtext[] = new byte[1];
        }


        class MyMsgBuf extends MsgBuf{
            public NativeLong messageKind;
            public byte[] contenido = new byte[1024];
        }

        // Initialize queue, or if it exists, get it            
        int msgget(NativeLong key, int msgflg);
        // Send messages to queue
        // int msgsnd(int msqid, struct msgbuf *ptrkey, int length, int flag);
        int msgsnd(int msqid, MsgBuf ptrkey, int msgsz, int msgflg);
        // Receive messages from queue
        // int msgrcv(int msqid, struct msgbuf *ptrkey, int length, long msgtype, int flag);
        int msgrcv(int msqid, MsgBuf ptrkey, int length, long msgtype, int flag);            

    }

    public static void main(String[] args) {

        int idCola = IPCLibrary.INSTANCE.msgget(new NativeLong(12500), 0);            

        if(idCola<0){
            System.out.println("The queue can't be created. IdCola:"+idCola);           
            System.out.println("Error msgget: " + Native.getLastError());
        }else{
            System.out.println("Queue with id:" + idCola + "has been recovered");           
            // Send message
            IPCLibrary.MyMsgBuf mensaje = new IPCLibrary.MyMsgBuf();
            mensaje.tipoMensaje = new NativeLong(1);
            mensaje.contenido = "Sending message".getBytes();
            int devSend = IPCLibrary.INSTANCE.msgsnd(idCola, mensaje, mensaje.contenido.length, 1);
            if(devSend != 0){
                System.out.println("Send response: "+devSend);
                System.out.println("Error value: " + Native.getLastError());
            }
        }

        // Receiving message
        IPCLibrary.MyMsgBuf mensajeRecibido =  new IPCLibrary.MyMsgBuf();        
        int bytesRecibidos = IPCLibrary.INSTANCE.msgrcv(idCola, mensajeRecibido, mensajeRecibido.contenido.length, 1234, 0);
        if(bytesRecibidos > 0){
            System.out.println("C message has been received: " + new String(mensajeRecibido.contenido));
        }else{
            System.out.println("msgrcv error: " + Native.getLastError());
        }


        // Send closing message
        IPCLibrary.MyMsgBuf mensajeCierre = new IPCLibrary.MyMsgBuf();
        mensajeCierre.tipoMensaje = new NativeLong(2);
        mensajeCierre.contenido = "Closing queue".getBytes();
        int devSend = IPCLibrary.INSTANCE.msgsnd(idCola, mensajeCierre, mensajeCierre.contenido.length, 1);
        if(devSend != 0){
            System.out.println("Send response: "+devSend);
            System.out.println("Error value: " + Native.getLastError());
        }            

    }
}

我真的希望这可以帮助别人。

Answer 2

我清理了你的代码并让它运行起来。 你需要两个任务。一发一收。 只需将上一篇文章中的 main() 替换为下面的 main()。它对我有用。感谢您发布您的努力让我开始。 另请参阅 kirk.c 和 spock.c 以获得良好的 c 示例http://beej.us/guide/bgipc/output/html/multipage/mq.html

   public static void main(String[] args) {
   double SLEEP_MINUTES = 0.1;
   int IPC_CREAT = 01000;  // starts with 0 so its octal or 512 
   int IPC_EXCL  = 02000;
   int IPC_NOWAIT  = 04000;
   int MSG_EXCEPT = 020000;
   int MSG_NOERROR =  010000;  // truncate the message if its to big     
   int msgflg_msgrcv = MSG_NOERROR; //  truncate the message if its to big
   int msgflg_msgget = 0666 | IPC_CREAT;
   int msgflg_msgsnd = 0;
   int msgtype_msgrcv = 0; // read them all
   NativeLong msgtype_msgsnd = new NativeLong(1); // just needs to be a positive number
   NativeLong msgkey = new NativeLong(12500);  



    int msqid = IPCLibrary.INSTANCE.msgget(msgkey, msgflg_msgget);            

    if(msqid<0)
    {
        System.out.println("The queue can't be created. msqid:"+msqid);           
        System.out.println("Error msgget: " + Native.getLastError());
        System.exit(0);
    }

    System.out.println("Queue with id:" + msqid + "has been found or created");  
    for(int i=0;i<100;i++)
    {
        // Send message
        IPCLibrary.MyMsgBuf message = new IPCLibrary.MyMsgBuf();
        message.messagetype = msgtype_msgsnd;
        message.content = ("Sending message"+i+'\0').getBytes();           // add 1 for the '\0'
        int devSend = IPCLibrary.INSTANCE.msgsnd(msqid, message, message.content.length+1,
                msgflg_msgsnd);
        if(devSend != 0)
        {
            System.out.println("Send response: "+devSend);
            System.out.println("Error value: " + Native.getLastError());
            System.exit(0);
        }
        System.out.println("Sent "+i);            
        try
        {
            Thread.sleep((long)(SLEEP_MINUTES*60.0*1000.0)); 
        }
        catch (InterruptedException e) 
        {
            System.out.println("InterruptedException while writing");
            System.out.println(e.getMessage());                 
        }

    }

}


     public static void main(String[] args) {
// found these in /usr/include/bits/*.h
int IPC_CREAT = 01000;  // remember if it starts with a '0' its octal or 512 
int IPC_EXCL  = 02000;
int IPC_NOWAIT  = 04000;
int MSG_EXCEPT = 020000;
int MSG_NOERROR =  010000;  // truncate the message if its to big     

int msgflg_msgrcv = MSG_NOERROR; //  truncate the message if its to big
int msgflg_msgget = 0666 | IPC_CREAT; // create the queue if its not there , let everybody read and write
int msgtype_msgrcv = 0; // read them all
NativeLong msgtype_msgsnd = new NativeLong(1); // just needs to be a positive number
NativeLong msgkey = new NativeLong(12500);  

    int msqid = IPCLibrary.INSTANCE.msgget(msgkey, msgflg_msgget);            
    if(msqid<0)
    {
        System.out.println("The queue can't be created. msqid:"+msqid);           
        System.out.println("Error msgget: " + Native.getLastError());
        System.exit(0);
    }

    System.out.println("Queue with id:" + msqid + "has been found or was created");  
    for(int i=0;i<100;i++)
    {
        IPCLibrary.MyMsgBuf message =  new IPCLibrary.MyMsgBuf();        
        int ret = IPCLibrary.INSTANCE.msgrcv(msqid, message, message.content.length,
                msgtype_msgrcv,msgflg_msgrcv);
        if(ret > 0)
        {
            System.out.println("message has been received: " + ret);
            System.out.println(new String(message.content));
        }
        else
        {
            System.out.println("msgrcv error: " + Native.getLastError());
        }
    }

   }
 }