Переход с Thread на Concurrent

KiraLis39 · Регистрация: 23.07.2014

Студворк — интернет-сервис помощи студентам

Сижу в раздумьях. Как и попытка перехода на JavaFX, попытка перейти на Concurrent вызвала лишь массу недоумений:

Я пишу, для обучения, практики, сетевой чат. Всё работает! При запуске Клиента запускается этот поток:

(breakerFlag - просто флаг, переключающийся перед закрытием программы, для верности; connected - переходит в true после успешного подключения к Серверу и обратно - при дисконнекте; onReceieveString(in.readUTF()); - вынесенный код, читающий из потока, т.е. in и out - ридер и врайтер из сокета сервера; Лонг was изначально равен 0. так что при первом входе он всегда меньше 30000 и идет попытка коннекта)

Java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
clientThread = new Thread(new Runnable() {
            @Override
            public void run() {
                // Пока не сбит флаг - работаем:
                while (!breakerFlag) {
                    
                    if (!connected) {
                        // Если подключение сорвалось или это первый проход - подключаемся.
                        if (System.currentTimeMillis() - was > 30000) {reConnect();}
                    } else {
                        // Если уже подключено - ждем сообщения:
                        try {onReceieveString(in.readUTF());} catch (Exception e) {onException(e);}
                    }
                    // Не перегружаем процессор, Даём место для отдыха и других потоков, если будут:
                    try {Thread.sleep(250);} catch (Exception e) {}
                }
            }
 
            private void reConnect() {
                try {
                    // запоминаем время и подключаемся к серверу:
                    was = System.currentTimeMillis();
                    jtArea.append("Подключение к серверу...\n");
                    fromserver = new Socket(ip, port);
                    onConnection();
                } catch (IOException e) {onDisconnect();}
            }
        });
        clientThread.setName("Сlient-" + nick + "`s thread");
        clientThread.start();

Поток без каких-либо проблем подключется к серверу, идет обмен сообщениями И при разрыве - переподключается через 30 секунд.
То-есть, даже если коннекта нет - по таймеру в 30 сек идет повтор. Итак!

Сегодня я подумал - уже время прошло, есть всеми любимый, удобный и востребованный пакет Concurrent!
Надо переписать код на него. Почитал, посмотрел примеры, сделал так:

Java
1
2
3
4
5
6
7
8
9
10
11
ExecutorService executor = Executors.newFixedThreadPool(3);
        executor.submit(() -> {
            if (breakerFlag) {executor.shutdownNow();}
            if (!connected) {
                // Если подключение сорвалось или это первый проход - подключаемся.
                if (System.currentTimeMillis() - was > 30000) {reConnect();}
            } else {
                // Если уже подключено - ждем сообщения:
                try {onReceieveString(in.readUTF());} catch (Exception e) {onException(e);}
            }
        });

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

Тогда я подумал о Семафорах. Так же использую впервые, чтоб их..
Посмотрел, почитал. Понял, что можно создать Конкурент-пул из нескольких потоков, а на трудный участок пустить только несколько, остальные может обойдут, как в оригинальном коде, проблемный участок и всетки начнут слежение за таймером..

Я переписал код с Семафором:

Java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
Semaphore semaphore = new Semaphore(1);
        ExecutorService executor = Executors.newFixedThreadPool(3);
        executor.submit(() -> {
            if (breakerFlag) {executor.shutdownNow();}
            if (!connected) {
                // Если подключение сорвалось или это первый проход - подключаемся.
                if (System.currentTimeMillis() - was > 30000) {reConnect();}
            } else {
                // Если уже подключено - ждем сообщения:
                try {
                    semaphore.acquire(1);
                    onReceieveString(in.readUTF());
                } catch (Exception e) {onException(e);
                } finally {semaphore.release();}
            }
        });

Что поменялось? Ничего.
При запуске без серва или дисконнекте - все так же намертво и навсегда вешается. Сообщения так же идут Только на сервер, в одну сторону.

Как я понял, теперь тупо все потоки стоят в очереди на коннект, который недоступен и некому следить за таймером переподключения, с которым великолепно справляется оригинальный thread.

И вот вопрос - что же это за чудесный Конкурент, используя который надо так дико выплясывать с бубнами? Подскажите, как все же НАДО правильно обрабатывать такие места, чтобы код корректно работал, как в оригинале?..

Еще раз:
В оригинале с Thread, если запустить Клиента без Сервера, но затем запустить Сервер - Клиент подключится сам к нему в течении следующих 30 сек, так как ищет его каждые 30 сек. В одном потоке.
В варианте с Конкуренси - даже включение Сервера уже никогда не вернет Клиента к жизни.

Благодарю заранее.

@KEKCoGEN · 12.11.2018, 01:28

Сообщение от KiraLis39

В варианте с Конкуренси - даже включение Сервера уже никогда не вернет Клиента к жизни.

что мешает пройтись дебагом или добавить логов чтобы посмотреть где проблема? По коду мало что понятно, но скорее всего изначально код написан не потокобезопасно поэтому когда появились потоки - начались проблемы. Например was она одна на все потоки чтоли?
Так же куда делся цикл в версии с потоками?

В целом очень похоже на программирование методом тыка...

KiraLis39 · 12.11.2018, 08:06 **[ТС]**

Так ведь.. я же не менял суть кода и логику..
Просто заменил тело Thread на тело Concurent-а

Проверю, устал просто пока. Посмотрю, конечно. Может просто были у кого уже такие моменты и были разобраны и выяснены.

Сообщение от KEKCoGEN

Так же куда делся цикл в версии с потоками?

Не понял, о чем речь. Какой цикл версии?..

Сообщение от KEKCoGEN

Например was она одна на все потоки чтоли?

was изначально равна нулю при инициализации, что вызывает сразу метод, где идет подключение к сокету серва. Затем, если булен connect, при дисконнекте, слетает, подключается метод подключения, где was сравнивается с текущим временем до 30 секунд, идет переподключение и сброс was. И так по кругу пока не подключится. Вот и все.

Добавлено через 3 минуты
Приведу полный код Основного класса Клиента, хоть проблема вроде и лишь в методе потока.. Но может поможет с определением проблемы:

(закомментирован старый блок, с которым все работает корректно)

Java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
package door;
 
import java.awt.BorderLayout;
import java.awt.Color;
import java.awt.Dimension;
import java.awt.event.KeyEvent;
import java.awt.event.KeyListener;
import java.awt.event.WindowEvent;
import java.awt.event.WindowListener;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.Socket;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
 
import javax.swing.JFrame;
import javax.swing.JOptionPane;
import javax.swing.JPanel;
import javax.swing.JScrollPane;
import javax.swing.JTextArea;
import javax.swing.JTextField;
import javax.swing.border.EmptyBorder;
import core.ConnectsClient;
 
/**
 * Не буду сильно углубляться уже во все детали: времени и так ушло много!
 * Пропускаю необязательные исключения и комменты - обрисую только ключенывае моменты пока:
 * 
 * Из метода {@link #main(String[])} сразу выходим, по привычке. В инициализацию, где:
 * либо берем данные для входа из {@value args[]}
 * либо предоставляем пользователю диалоговые окна для их ввода.
 * 
 * И создаем экземпляр класса {@link #Client(String)}, передавая имя клиента в него...
 * 
 * @author KiraLis39
 **/
public class Client implements ConnectsClient, WindowListener {
    private static String ip, nick;
    private static int port = 0;
    private static Random rand = new Random();
    private Socket fromserver;
    private DataInputStream in;
    private DataOutputStream out;
    private JTextArea jtArea;
    private JTextField inputField;
    private Thread clientThread;
    private boolean breakerFlag = false, connected = false;
    private long was = 0L;
    private JFrame clientTestFrame;
    
    private ExecutorService executor;
    private Semaphore semaphore;
    
    
    @SuppressWarnings("serial")
    public Client(String name) {
        // Фрейм GUI Клиента:
        clientTestFrame = new JFrame();
        clientTestFrame.setTitle(nick);
        clientTestFrame.setDefaultCloseOperation(JFrame.EXIT_ON_CLOSE);
        clientTestFrame.setPreferredSize(new Dimension(400, 400));
        clientTestFrame.addWindowListener(this);
        clientTestFrame.setAlwaysOnTop(true);
        
            // Базовая, задняя панель, куда будет крепиться все прочее:
            JPanel messageBaseTextPane = new JPanel(new BorderLayout()) {
                {
                    setBorder(new EmptyBorder(3, 3, 3, 3));
                }
            };
            
                // Большое текстовое поле для отображения сообщений:
                jtArea = new JTextArea() {
                    {
                        setBorder(new EmptyBorder(5, 5, 5, 5));
                        setEditable(false);
                        setLineWrap(true);
                        setWrapStyleWord(true);
                        
                        setBackground(Color.DARK_GRAY);
                        setForeground(Color.GREEN.brighter());
                    }
                };
                // Поле ввода нового сообщения для отправки:
                inputField = new JTextField() {
                    {
                        setBorder(new EmptyBorder(5, 5, 5, 5));
                        
                        setBackground(Color.BLACK);
                        setForeground(Color.GREEN.brighter());
                        
                        addKeyListener(new KeyListener() {
                            public void keyTyped(KeyEvent e) {}
                            public void keyReleased(KeyEvent e) {}
                            
                            @Override
                            public void keyPressed(KeyEvent e) {
                                if (e.getKeyCode() == KeyEvent.VK_ENTER) {onSendMessage(inputField.getText());}
                            }
                        });
                    }
                };
            
            messageBaseTextPane.add(new JScrollPane(jtArea), BorderLayout.CENTER);
            messageBaseTextPane.add(inputField, BorderLayout.SOUTH);
        
        clientTestFrame.add(messageBaseTextPane);       
        
        clientTestFrame.pack();
        clientTestFrame.setLocationRelativeTo(null);
        clientTestFrame.setVisible(true);
        
        start();
    }
        
    private void start() {
        semaphore = new Semaphore(1);
        executor = Executors.newFixedThreadPool(3);
        executor.submit(() -> {
            if (breakerFlag) {executor.shutdownNow();}
            if (!connected) {
                // Если подключение сорвалось или это первый проход - подключаемся.
                if (System.currentTimeMillis() - was > 30000) {reConnect();}
            } else {
                // Если уже подключено - ждем сообщения:
                try {
                    semaphore.acquire(1);
                    onReceieveString(in.readUTF());
                } catch (Exception e) {onException(e);
                } finally {semaphore.release();}
                
            Thread.yield();
            }
        });
        
        // Запуск основного потока обработки операции Клиента:
//      clientThread = new Thread(new Runnable() {
//          @Override
//          public void run() {
//              // Пока не сбит флаг - работаем:
//              while (!breakerFlag) {
//                  
//                  if (!connected) {
//                      // Если подключение сорвалось или это первый проход - подключаемся.
//                      if (System.currentTimeMillis() - was > 30000) {reConnect();}
//                  } else {
//                      // Если уже подключено - ждем сообщения:
//                      try {onReceieveString(in.readUTF());} catch (Exception e) {onException(e);}
//                  }
//                  // Не перегружаем процессор, Даём место для отдыха и других потоков, если будут:
//                  try {Thread.sleep(250);} catch (Exception e) {}
//              }
//          }
//
//          private void reConnect() {
//              try {
//                  // запоминаем время и подключаемся к серверу:
//                  was = System.currentTimeMillis();
//                  jtArea.append("Подключение к серверу...\n");
//                  fromserver = new Socket(ip, port);
//                  onConnection();
//              } catch (IOException e) {onDisconnect();}
//          }
//      });
//      clientThread.setName("Сlient-" + nick + "`s thread");
//      clientThread.start();
    }
    
    private void reConnect() {
        try {
            // запоминаем время и подключаемся к серверу:
            was = System.currentTimeMillis();
            jtArea.append("Подключение к серверу...\n");
            fromserver = new Socket(ip, port);
            onConnection();
        } catch (IOException e) {onDisconnect();}
    }   
 
    @Override
    public void onReceieveString(String message) {
        // Если пришла данная строка - следует отключиться:
        if (message.equals("DISCONNECT")) {
            onDisconnect();
            return;
        }
        
        if (message.equals("NAME_IS")) {
            try {
                out.writeUTF(getName());
                out.flush();
            } catch (Exception e) {jtArea.append("Name sended." + "\n");}
            return;
        }
        
        // иначе, просто выводим сообщение в GUI, в данном случае:
        String[] messageCuts = message.split(" ");
        
        if (messageCuts.length > 1) {
            if (messageCuts[0].endsWith(":")) {jtArea.append(message + "\n");} else {jtArea.append("UNCKNOWN: " + message + "\n");}
        } else {jtArea.append("SERVER: " + message + "\n");}
    }
    
    @Override
    public void onSendMessage(String text) {
        // трассировка исходящих сообщений (можно удалить, если не требуется):
        System.err.println("CLIENT >> " + text);
        
        // Если введена команда на выход - отключаемся:
        if (text.equalsIgnoreCase("quit")) {
            jtArea.append("Отключение от сервера...\n");
            onDisconnect();
            jtArea.append("Выход из программы...\n");
            onExit();
        }
        
        // в обычной ситуации отправляем письмо Серверу и дублируем в GUI:
        try {
            out.writeUTF(text);
            out.flush();
            
            jtArea.append(text + "\n");
            inputField.setText("");
        } catch (Exception e) {jtArea.append("Sending failed." + "\n");}
 
        inputField.requestFocus();
    }
 
    @Override
    public void onConnection() {
        System.out.println("Client -> clientThread -> reConnect:: Connect to ip: " + ip + " and port: " + port + " with name '" + nick + "'.");
 
        // при подключении создаем каналы входа и выхода:
        try {
            in      = new DataInputStream(fromserver.getInputStream());
            out     = new DataOutputStream(fromserver.getOutputStream());
        } catch (IOException e) {
            e.printStackTrace();
            onDisconnect();
        }
        
        try {
            // сразу отправляем имя клиента, для идентификации:
            out.writeUTF(nick);
            out.flush();
        } catch (IOException e) {
            e.printStackTrace();
            onDisconnect();
        }
        
        connected = true;
        jtArea.append("Подключение к серверу выполнено!\n");
    }
 
    @Override
    public void onDisconnect() {
        if (!connected) {return;} // запасная заглушка для предотвращения случайных проскоков повтора.
 
        // если дисконнект спровоцирован местными событиями - можно на всякий случай сообщить Серверу что мы отключаемся.
        try {
            out.writeUTF("BYE_BYE");
            out.flush();
        } catch (Exception e) {onException(e);}
        
        breakerFlag = true;
        connected = false;
        
        try {clientTestFrame.dispose();} catch (Exception e) {/* обычное дело. Здесь ничего не требуется обычно.*/}
        try {clientThread.interrupt();} catch (Exception e) {/* обычное дело. Здесь ничего не требуется обычно.*/}
        try {out.close();} catch (Exception e) {/* обычное дело. Здесь ничего не требуется обычно.*/}
        try {in.close();} catch (Exception e) {/* обычное дело. Здесь ничего не требуется обычно.*/}
        try {fromserver.close();} catch (IOException e) {onException(e);}
    }
 
    private void onExit() {System.exit(0);}
    
    @Override
    public void onException(Exception e) {
        System.out.println("*** *** *** *** *** ***");
        System.out.println(e.getLocalizedMessage() + ": ");
        System.out.println("*** *** *** *** *** ***");
        e.printStackTrace(System.out);
    }
 
    @Override
    public String getName() {return nick;}
 
    
    public static void main(String[] args) {init(args);} //нет смысла задерживаться в точке входа.
    
    private static void init(String[] args) {
        try {
            if (args == null || args.length < 3) {
                // предлагаем пользователю ввести данные через диалог:
                ip = JOptionPane.showInputDialog(null, "Введи IP сервера", "127.0.0.1");
                port = Integer.valueOf( JOptionPane.showInputDialog(null, "Введи порт сервера", "8082"));
                nick = JOptionPane.showInputDialog(null, "Введи свой ник", "User_" + String.format("%.2f", rand.nextFloat() * 10));
            } else {
                // берем данные для работы из 'args[]':
                ip = args[0];
                port = Integer.valueOf(args[1]);
                nick = args[2];
            }
            
            // если введено не верно - выходим! Для начала достаточно и этого сейчас.
            if (ip == null || nick == null || ip.equals("") || nick.equals("")) {
                System.out.println("Данные введены не верно!");
                System.out.println("Если входите через cmd: 'ip_address' 'port' 'your_nickname'");
                System.exit(-1);
            }
            
            // Очевидно, все проверки пройдены - создаем экземпляр Клиента:
            new Client(nick);
        } catch (Exception e) {System.exit(-1);}
    }
    
    @Override // когда окно клиента закрывается - отключаемся от Сервера.
    public void windowClosing(WindowEvent e) {onDisconnect();}
    
    public void windowClosed(WindowEvent e) {}  
    public void windowOpened(WindowEvent e) {}
    public void windowIconified(WindowEvent e) {}
    public void windowDeiconified(WindowEvent e) {}
    public void windowActivated(WindowEvent e) {}
    public void windowDeactivated(WindowEvent e) {}
}

@korvin_ · 12.11.2018, 09:03

Сообщение от KiraLis39

И вот вопрос - что же это за чудесный Конкурент, используя который надо так дико выплясывать с бубнами?

Конкарент — нормальный, плясать там не надо, а надо почитать документацию, книжку какую-нибудь, а не «тяп-ляп и в продакшн».

Сообщение от KiraLis39

Так ведь.. я же не менял суть кода и логику..

Где же ты её не менял, если в первом коде у тебя while (...), а во втором — if?

@KEKCoGEN · 12.11.2018, 09:53

KiraLis39, ваще жесть какая-то написана. Похоже ты просто не понимаешь тему конкаренси. Попробуй простых примеров пописать для начала. У тебя переменная was общая для всех потоков. Подумай как себя будет вести первый и последующие потоки.

@korvin_ · 12.11.2018, 10:10

Сообщение от KEKCoGEN

У тебя переменная was общая для всех потоков. Подумай как себя будет вести первый и последующие потоки.

По факту у него там только один поток, но да, всё через гонки — это жесть. Кроме того, мне не понятно, как swing позволяет ему вызывать методы UI-компонентов из другого потока, насколько помню, он на такое бросает исключение.

@KEKCoGEN · 12.11.2018, 10:28

Сообщение от korvin_

насколько помню, он на такое бросает исключение.

вроде андроид бросает. В свинге это делать можно, но может подвисать если invokeLater не и спользовать.

KiraLis39 · 12.11.2018, 15:24 **[ТС]**

Сообщение от korvin_

мне не понятно, как swing позволяет ему вызывать методы UI-компонентов из другого потока, насколько помню, он на такое бросает исключение.

Это привилегии тех, кто идет своим путем, не только копируя других )

Добавлено через 2 минуты

Сообщение от KEKCoGEN

Попробуй простых примеров пописать для начала.

Пожалуй.. Думал, это будет проще.
Видимо, нет здесь готовых, простых решений. Что поделать..

@korvin_ · 12.11.2018, 16:15

Сообщение от KiraLis39

Это привилегии тех, кто идет своим путем, не только копируя других )

Привелегии писать говнокод? Окай.

Сообщение от KiraLis39

Видимо, нет здесь готовых, простых решений. Что поделать..

Новые блоги и статьи Все статьи Все блоги /
SDL3 для Web (WebAssembly): Загрузка PNG с прозрачным фоном с помощью SDL3_image 8Observer8 10.02.2026 Содержание блога Библиотека SDL3_image содержит инструменты для расширенной работы с изображениями. Пошагово создадим проект для загрузки изображения формата PNG с альфа-каналом (с прозрачным. . .	Установка Qt-версии Lazarus IDE в Debian Trixie Xfce volvo 10.02.2026 В общем, достали меня глюки IDE Лазаруса, собранной с использованием набора виджетов Gtk2 (конкретно: если набирать текст в редакторе и вызвать подсказку через Ctrl+Space, то после закрытия окошка. . .	SDL3 для Web (WebAssembly): Работа со звуком через SDL3_mixer 8Observer8 08.02.2026 Содержание блога Пошагово создадим проект для загрузки звукового файла и воспроизведения звука с помощью библиотеки SDL3_mixer. Звук будет воспроизводиться по клику мышки по холсту на Desktop и по. . .	SDL3 для Web (WebAssembly): Основы отладки веб-приложений на SDL3 по USB и Wi-Fi, запущенных в браузере мобильных устройств 8Observer8 07.02.2026 Содержание блога Браузер Chrome имеет средства для отладки мобильных веб-приложений по USB. В этой пошаговой инструкции ограничимся работой с консолью. Вывод в консоль - это часть процесса. . .
SDL3 для Web (WebAssembly): Обработчик клика мыши в браузере ПК и касания экрана в браузере на мобильном устройстве 8Observer8 02.02.2026 Содержание блога Для начала пошагово создадим рабочий пример для подготовки к экспериментам в браузере ПК и в браузере мобильного устройства. Потом напишем обработчик клика мыши и обработчик. . .	Философия технологии iceja 01.02.2026 На мой взгляд у человека в технических проектах остается роль генерального директора. Все остальное нейронки делают уже лучше человека. Они не могут нести предпринимательские риски, не могут. . .	SDL3 для Web (WebAssembly): Вывод текста со шрифтом TTF с помощью SDL3_ttf 8Observer8 01.02.2026 Содержание блога В этой пошаговой инструкции создадим с нуля веб-приложение, которое выводит текст в окне браузера. Запустим на Android на локальном сервере. Загрузим Release на бесплатный. . .	SDL3 для Web (WebAssembly): Сборка C/C++ проекта из консоли 8Observer8 30.01.2026 Содержание блога Если вы откроете примеры для начинающих на официальном репозитории SDL3 в папке: examples, то вы увидите, что все примеры используют следующие четыре обязательные функции, а. . .

@KEKCoGEN 2399 / 2224 / 565 Регистрация: 28.12.2010 Сообщений: 8,672
	12.11.2018, 09:53
	Сообщение было отмечено KiraLis39 как решение Решение KiraLis39, ваще жесть какая-то написана. Похоже ты просто не понимаешь тему конкаренси. Попробуй простых примеров пописать для начала. У тебя переменная was общая для всех потоков. Подумай как себя будет вести первый и последующие потоки. 1

Переход с Thread на Concurrent

Решение