5

次の図に示すように、XPUBとXSUBを使用して実装しようとしています。提供されている例を確認しましたが、JavaのXPUBおよびXSUB用の例を取得できませんでした。ここでは、私がZeroMQを初めて使用するため、少し複雑なCの例を示しています。

ZeroMQのプロキシを使用したPub-Subネットワーク
jniラップバージョンを使用してAndroidで使用しようとしています。Javaを使用してZeroMQでプロキシを使用してこのPub-Subネットワークを実装する方法の例を見つけるのを手伝ってください。

現在、私はhttp://zguide.zeromq.org/page:allを参照しています

私はそれを次のように移植しようとしました。 Subscriber.java

パブリッククラスサブスクライバーはスレッドを拡張し、Runnableを実装します{



private static final String TAG = "Subscriber";
private Context ctx;

public Subscriber(ZMQ.Context z_context) {
    this.ctx = z_context;
}

@Override
public void run() {

    super.run();

    ZMQ.Socket mulServiceSubscriber = ctx.socket(ZMQ.SUB);
    mulServiceSubscriber.connect("tcp://localhost:6001");
    mulServiceSubscriber.subscribe("A".getBytes());
    mulServiceSubscriber.subscribe("B".getBytes()); 


        while (true) {
            Log.d(TAG, "Subscriber loop started..");
            String content = new String(mulServiceSubscriber.recv(0));
            Log.d(TAG, "Subscriber Received : "+content);
        }
}




}

Publisher.java 

public class Publisher extends Threadimplements Runnable {



private static final String TAG = "Publisher";
private Context ctx;

public Publisher(ZMQ.Context z_context) {
    this.ctx = z_context;
}

@Override
public void run() {

    super.run();

    ZMQ.Socket publisher = ctx.socket(ZMQ.PUB);
    publisher.connect("tcp://localhost:6000");

    while (true) {
        Log.d(TAG, "Publisher loop started..");
        publisher.send(("A Hello " + new Random(100).nextInt()).getBytes() , 0);
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}




}

XListener.java（今のところ単純なフォワーダー）

パブリッククラスXListenerはスレッドを拡張し、Runnableを実装します{



private static final String TAG = null;
private Socket publisherX;
private Context ctx;
private Socket subscriberX;

public XListener(ZMQ.Context ctx, ZMQ.Socket subscriberX,
        ZMQ.Socket publisherX) {
    this.ctx = ctx;
    this.subscriberX = subscriberX;
    this.publisherX = publisherX;

}

@Override
public void run() {
    super.run();
    while (true) {          
        Log.d(TAG, "XListener loop started..");

        String msg = new String(subscriberX.recvStr());
        Log.v(TAG, "Listener Received: " +"MSG :"+msg);
        publisherX.send(msg.getBytes(), 0);         
    }
}




}

アプリケーションmain（）で

private void main（）{
        ZMQ.Context ctx = ZMQ.context（1）;   


    ZMQ.Socket subscriberX = ctx.socket(ZMQ.XSUB);
    subscriberX.bind("tcp://*:6000");
    try {
        Thread.sleep(500);
    } catch (InterruptedException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
    ZMQ.Socket publisherX = ctx.socket(ZMQ.XPUB);
    publisherX.bind("tcp://*:6001");
    try {
        Thread.sleep(500);
    } catch (InterruptedException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }

    new XListener(ctx, subscriberX, publisherX).start();
    try {
        Thread.sleep(500);
    } catch (InterruptedException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }

   new XSender(ctx, subscriberX, publisherX).start();
    try {
        Thread.sleep(500);
    } catch (InterruptedException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
    new Subscriber(ctx).start();
    new Publisher(ctx).start();
}

コードでは、XSUBを聞くことができません。espresso.cを移植しているときに、ZMQのJavaバインディングでラッパーを見つけることができませんでした。単純なプロキシを実装する方法、または何かが足りないのですか？

4

1 に答える 1

1

うわー、私は自分の質問に答えています。PublisherXからsubscriberXへのフォワーダーを追加できませんでした。不足しているコードは次のとおりです。これで、XSUBとXPUBはデータを送受信できるようになります。

パブリッククラスXSenderはスレッドを拡張し、Runnableを実装します{



private static final String TAG = null;
private Socket publisherX;
private Context ctx;
private Socket subscriberX;

public XSender(ZMQ.Context ctx, ZMQ.Socket subscriberX,
        ZMQ.Socket publisherX) {
    this.ctx = ctx;
    this.subscriberX = subscriberX;
    this.publisherX = publisherX;

}

@Override
public void run() {
    super.run();
    while (true) {
        // Read envelope with address
        Log.d(TAG, "XListener loop started..");

        String msg = new String(subscriberX.recv(0));
        Log.v(TAG, "Listener Received: " +"MSG :"+msg);
        publisherX.send(msg.getBytes(), 0);         

    }


}




}

于 2013-01-28T08:46:27.393 に答える