Communication Manager, который контролирует скорость асинхронного сетевого обмена сообщениями - PullRequest
Новая
       88

Communication Manager, который контролирует скорость асинхронного сетевого обмена сообщениями

0 голосов
/ 21 февраля 2019

У меня есть система, к которой мне нужно получить доступ, и скорость доступа к этой системе составляет 1 вызов API в секунду.Тем не менее, я хочу предоставить доступ к нему через веб-интерфейс, который является асинхронным.У меня есть проект, который вызывает выделенный поток менеджера связи, который собирает сообщения в очередь и передает их по одному, а затем отправляет обратный вызов отправителю с результатами сообщения.

Этохороший подход?Видите ли вы какие-либо очевидные подводные камни в моем коде в его текущем состоянии? 

public class CommunicationManager implements Runnable
{
    private BlockingQueue<Message> message = new LinkedList<> ();

    private boolean shutdown = false;

    private Messenger messenger = new Messenger();

    public CommunicationManager() {}

    public void run()
    {
        long elapsed, start, diff;
        start = 0;

        while (!shutdown)
        {
            elapsed = System.currentTimeMillis();
            diff = elapsed - start;
            if (diff < 1000)
            {
                Thread.sleep(1000 - diff);
            }
            if (!message.isEmpty())
            {
                Message next = message.remove();
                next.getSender().recieve(messenger.send(next.getMessage()));
            }
            start = elapsed;
        }           
    }

    public synchronized void addMessage(Sender sender, String message)
    {
        this.message.add(new Message(sender, message));
    }

    public synchronized void shutdown()
    {
        this.shutdown = true;
    }
}

Обнадеживающим результатом этого менеджера является то, что каждый цикл, если секунда не прошла с момента запуска предыдущего цикла,будет спать в течение оставшегося времени.Затем он проверит, пуста ли очередь.Если очередь не пуста, она извлечет следующее сообщение в очереди, отправит сообщение и вернет результат обратному вызову отправителя.Затем цикл заканчивается, и он снова начинает цикл.

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

1 Ответ

0 голосов
/ 02 марта 2019

Оказывается, я немного заново изобрел колесо.То, что я действительно хотел сделать, можно сделать намного проще с помощью Java ScheduledExecutorService. 

class CommunicationControl
{
   private final ScheduledExecutorService scheduluer = Executors.newScheduledThreadPool(1);

   public void startManager()
   {
      final CommunicationManager manager = new CommunicationManager();
      scheduler.scheduleAtFixedRate(manager, 10, 1, SECONDS);
   }

   public void stopManager()
   {
      while (manager.getMessageCount() > 0)
      {
          try
          {
              Thread.sleep(manager.getMessageCount() * 1000);
          } catch (InterruptedException e)
          {
              e.printStackTrace();
          }
      }
      scheduler.shutdown();
   }
}

Переработка исходного класса здесь: 

class CommunicationManager implements Runnable
{
   private BlockingQueue<Message> message = new ArrayBlockingQueue<Message> (1000);
   private Messenger messenger = new Messenger();

   public CommunicationManager() {}

   public void run()
   {
      Message next = message.poll();
      if (next != null)
      {
         next.getSender().recieve(messenger.send(next.getMessage()));
      }
   }

   public void addMessage(Sender sender, String message)
   {
       try
       {
          while (!this.message.offer(new Message(sender, message), 1, TimeUnit.SECONDS)) {}
       } catch (InterruptedException e)
       {
          e.printStackTrace();
       }
   }
}

Что это будет делать, это сначала создатьисполнитель для управления частотой запуска моего Communication Manager, который в этом случае выполняется один раз в секунду с задержкой в ​​10 секунд.Затем, как только это происходит каждую секунду, метод Run CommuncationManager будет выполнять, выполняя ровно одну единицу работы, если она доступна.

Другие методы могут добавлять работу в CommunicationManager, если есть место в очереди, в противном случае они будутблокировать, пока пространство не станет доступным.

Это также освобождает меня от возможности встроить это в среду, такую ​​как Spring.

...