2 つのアプリケーションがあり、双方向に非同期でメッセージを送信しています。両側で ZMQ.DEALER タイプのソケットを使用しています。接続ステータスは、さらにハートビートによって制御されます。
接続の問題（回線障害または一方のアプリケーションの再起動）の後、接続を確実に回復するのに問題があります。サーバー側 (bind() を実行する側) でアプリケーションを再起動すると、クライアント側は常に正常に再接続するとは限らず、特にローカル バッファーが HWM 制限に達した場合に再起動する必要があります。
ハートビート障害が発生した場合、または send() が false を返した場合に完全な ZMQ.Context をリセットする以外に、接続回復を信頼できるものにする他の方法は見つかりませんでした。次に、Context.term() を呼び出して、Context と Socket を再度作成します。これは私のテストではうまくいくようでした。しかし今、Context.term() 内で時折発生するハングアップを観察しましたが、これはまれであり、再現が困難です。Context の作成は、アプリケーションの起動時に一度だけ行う必要があることはわかっていますが、前述のように、切断された接続を再確立する他の方法は見つかりませんでした。
JeroMQ 0.3.4 を使用しています。テスト アプリケーションのソースは次のとおりです。約 200 行のコードです。
これを解決するためのヒントは大歓迎です。
import java.util.Calendar;
import org.zeromq.ZMQ;
public class JeroMQTest {
public interface IMsgListener {
public void newMsg(byte[] message);
}
final static int delay = 100;
final static boolean doResetContext = true;
static JeroMQTest jeroMQTest;
static boolean isServer;
private ZMQ.Context zContext;
private ZMQ.Socket zSocket;
private String address = "tcp://localhost:9889";
private long lastHeartbeatReceived = 0;
private long lastHeartbeatReplyReceived;
private boolean sendStat = true, serverIsActive = false, receiverInterrupted = false;
private Thread receiverThread;
private IMsgListener msgListener;
public static void main(String[] args) {
isServer = args.length > 0 && args[0].equals("true");
if (isServer) {
new JeroMQTest().runServer();
}
else {
new JeroMQTest().runClient();
}
}
public void runServer() {
msgListener = new IMsgListener() {
public void newMsg(byte[] message) {
String msgReceived = new String(message);
if (msgReceived.startsWith("HEARTBEAT")) {
String msgSent = "HEARTBEAT_REP " + msgReceived.substring(10);
sendStat = zSocket.send(msgSent.getBytes());
System.out.println("heartbeat rcvd, reply sent, status:" + sendStat);
lastHeartbeatReceived = getNow();
} else {
System.out.println("msg received:" + msgReceived);
}
}
};
createJmq();
sleep(1000);
int ct = 1;
while (true) {
boolean heartbeatsOk = lastHeartbeatReceived > getNow() - delay * 4;
if (heartbeatsOk) {
serverIsActive = true;
String msg = "SERVER " + ct;
sendStat = zSocket.send(msg.getBytes());
System.out.println("msg sent:" + msg + ", status:" + sendStat);
ct++;
}
if (serverIsActive && (!heartbeatsOk || !sendStat)) {
serverIsActive = false;
if (doResetContext) {
resetContext();
}
}
sleep(delay);
}
}
public void runClient() {
msgListener = new IMsgListener() {
public void newMsg(byte[] message) {
String msgReceived = new String(message);
if (msgReceived.startsWith("HEARTBEAT_REP")) {
System.out.println("HEARTBEAT_REP received:" + msgReceived);
lastHeartbeatReplyReceived = getNow();
}
else {
System.out.println("msg received:" + msgReceived);
}
}
};
createJmq();
sleep(1000);
int ct = 1;
boolean reconnectDone = false;
while (true) {
boolean heartbeatsOK = lastHeartbeatReplyReceived > getNow() - delay * 4;
String msg = "HEARTBEAT " + (ct++);
sendStat = zSocket.send(msg.getBytes());
System.out.println("heartbeat sent:" + msg + ", status:" + sendStat);
sleep(delay / 2);
if (sendStat) {
msg = "MSG " + ct;
sendStat = zSocket.send(msg.getBytes());
System.out.println("msg sent:" + msg + ", status:" + sendStat);
reconnectDone = false;
}
if ((!heartbeatsOK && lastHeartbeatReplyReceived > 0) || (!sendStat && !reconnectDone)) {
if (doResetContext) {
resetContext();
}
lastHeartbeatReplyReceived = 0;
reconnectDone = true;
}
sleep(delay / 2);
}
}
public void resetContext() {
closeJmq();
sleep(1000);
createJmq();
System.out.println("resetContext done");
}
private void createJmq() {
zContext = ZMQ.context(1);
zSocket = zContext.socket(ZMQ.DEALER);
zSocket.setSendTimeOut(100);
zSocket.setReceiveTimeOut(100);
zSocket.setSndHWM(10);
zSocket.setRcvHWM(10);
zSocket.setLinger(100);
if (isServer) {
zSocket.bind(address);
} else {
zSocket.connect(address);
}
receiverThread = new Thread() {
public void run() {
receiverInterrupted = false;
try {
ZMQ.Poller poller = new ZMQ.Poller(1);
poller.register(zSocket, ZMQ.Poller.POLLIN);
while (!receiverInterrupted) {
if (poller.poll(100) > 0) {
byte byteArr[] = zSocket.recv(0);
msgListener.newMsg(byteArr);
}
}
poller.unregister(zSocket);
} catch (Throwable e) {
System.out.println("Exception in ReceiverThread.run:" + e.getMessage());
}
}
};
receiverThread.start();
}
public void closeJmq() {
receiverInterrupted = true;
sleep(100);
zSocket.close();
zContext.term();
}
long getNow() {
Calendar now = Calendar.getInstance();
return (long) (now.getTime().getTime());
}
private static void sleep(int mSleep) {
try {
Thread.sleep(mSleep);
} catch (InterruptedException e) {
}
}
}