java сетевое программирование координация обмена сообщениями

Question

У меня есть 2 процесса, выполняющиеся на разных машинах, которые обмениваются данными через TCP-сокеты.
Оба процесса имеют код, который действует как сервер, так и как клиент.
I.e. ProcessA открыл сокет сервера, который связывается с портомX и ProcessB открыл привязку сокета к серверу portY.
ProcessA открывает клиентский сокет для соединения с ProcessB и начинает отправлять сообщения как клиент и получает ответы (по одному и тому же соединению с tcp, конечно).
ProcessB, когда он получает сообщение и обрабатывает его, он отправляет ответ, но также может отправлять сообщение по второму подключению tcp, то есть где ProcessB открыл клиентский сокет для порта X ProcessA.
Таким образом, поток сообщений превышает 2 различных соединения tcp.
Моя проблема заключается в следующем: Принимая, как должное, что эта «архитектура» не может измениться, и должны оставаться как есть:
У меня есть проблема, что периодически, сообщения посылать из ProcessB в ProcessA через соединение TCP, что ProcessB имеет открыл клиентский сокет, дошел до процесса А до того, как сообщения отправили в качестве ответов от ProcessB на ProcessA по подключению tcp, которое ProcessA подключил как клиентский сокет.
I.e. Оба потока происходит

(1) 
ProcessA ---->(msg)----> ProcessB(PortY) (TCP1) 
ProcessB does processing 
ProcessB(portY)--->(response)----->ProcessA (TCP1) 
ProcessB--->(msg)----->ProcessA(portX) (TCP2) 

(2) 
ProcessA ---->(msg)----> ProcessB(PortY) (TCP1) 
ProcessB does processing 
ProcessB--->(msg)----->ProcessA(portX) (TCP2) 
ProcessB(portY)--->(response)----->ProcessA (TCP1) 

EDIT (после запроса EJP) Как я могу применять/убедитесь, что ProcessB не посылает Сообщ через соединение, что ProcessB имеет сокет клиента, открытый для сервера PortX из ProcessA до сообщение посылает как ответ от сервера portY процесса ProcessB приходит к processA? То есть иметь только поток (1) из приведенного выше.
Обратите внимание, что processB многопоточен, а обработка является нетривиальной.

UPDATE: Может быть, это мое заблуждение, но когда процесс отправляет данные через сокет, и управление возвращается к применению, это не означает, что принимающая сторона получает данные. Итак, если процесс отправляет данные через 2 сокета, есть ли состояние гонки по ОС?

UPDATE2
После ответа я получил от Виджай Mathew:
Если бы я сделал замок, как предложено, есть гарантия того, что ОС (т.е. слой IP) будет передавать данные в порядке? То есть закончить одну передачу, а затем отправить следующий? Или я бы их мультиплексировал и имел такую ​​же проблему?

Благодаря

Answer 1

очевидное решение:

LockObject lock; 

ProcessA ---->(msg)----> ProcessB(PortY) 

// Processing the request and sending its response 
// makes a single transaction. 
synchronized (lock) { 
    ProcessB does processing 
    ProcessB(portY)--->(response)----->ProcessA (TCP1) 
} 

// While the processing code holds the lock, B is not 
// allowed to send a request to A. 
synchronized (lock) { 
    ProcessB--->(msg)----->ProcessA(portX) (TCP2) 
} 

Answer 2

Очевидный вопрос почему вы заботитесь? Если у вас есть операции, которые необходимо синхронизировать с обоих концов, сделайте это. Не ожидайте, что TCP сделает это за вас, это не то, для чего.

Answer 3

Проблема с синхронизацией может быть не в протоколе TCP, а в обработчике потока, который выбирает, какой поток будет просыпаться при поступлении сообщений. По характеру вашего вопроса я понимаю, что PortX «(Msg)» отправляется очень быстро после «PortY» (Response) ». Это означает, что обработчик потока иногда может выбирать, какой из двух прослушивающих потоков он будет просыпаться.

Простой, но уродливый и неполный способ устранения проблемы заключается в вставке короткого сна между ответом и следующим сообщением. Сон должен быть достаточно длинным, чтобы быть уверенным, что другой процесс проснется до ответа до получения следующего сообщения. Этот путь является неполным, потому что, хотя вы увеличиваете изменения в правильной синхронизации вашей обработки, такие проблемы, как загрузка ОС и перегрузка сети, всегда могут сдержать, чтобы ваше сообщение вернулось за ваш ответ. И тогда ты вернулся туда, где начал. Другой кусочек уродства заключается в том, что время сна сокращается и уменьшает максимальную пропускную способность. Реже реже. Итак ...

Чтобы полностью решить проблему, необходимо, чтобы каждый слушатель сокета нуждался в каком-либо способе узнать, является ли только что полученное сообщение следующей обработкой или могут быть более ранние сообщения, которые должны быть обработаны первый. Сделайте это путем последовательной нумерации всех сообщений, отправленных каждым процессом. Тогда процесс получения знает, что чего-то не хватает.

Вам нужно будет подумать о том, как слушатели в каждом соке могут общаться между собой, чтобы гарантировать, что полученные сообщения обрабатываются в порядке передачи. Существует ряд практических решений, но все они одинаковы на абстрактном концептуальном уровне.

THREAD 1: A) Образец ProcessA (PortX) получает сообщение и просыпается.
B) ЕСЛИ порядковый номер указывает, что отсутствует сообщение, THEN B1) синхронизироваться на ProcessA (PortY) и ждать(). B2) При пробуждении, назад к B) C) {сообщение отсутствует] Обработать сообщение. D) Назад к А)

РЕЗЬБА 2: A) ProcessA (PortY) получает ответ и просыпается. B) Обработать ответ. C) notifyAll(). D) Назад к A)

Наиболее общие практические решения, вероятно, будут включать в себя один экземпляр прослушивателя сокетов, добавляющий все новые сообщения в PriorityQueue, поэтому самые ранние отправленные сообщения всегда идут в очередь очереди. Тогда Thread 1 и Thread 2 могли ждать в этом экземпляре до тех пор, пока не поступит сообщение, которое они могут обработать.

Простейшее, но менее расширяемое решение состоит в том, чтобы каждый поток выполнял собственное прослушивание и ожидание с обработчиком (ответчиком), уведомляющим после обработки.

Удача в этом, хотя после всего этого времени это, вероятно, уже решено.