zhubaomin
2025-02-25 842237345ac469b02e9add8f9fd8bae5d4f7cdac
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
package com.dy.rtuMw.server.forTcp;
 
import java.util.*;
import java.util.Map.Entry;
 
import com.dy.common.mw.channel.tcp.TcpIoSessionAttrIdIsRtuAddr;
import org.apache.mina.core.session.IoSession;
 
import com.dy.rtuMw.server.ServerProperties;
 
 
public class TcpSessionCache {
    
    /**
     * 用Hashtable而不用HashMap原因:
     * Hashtable线程安全的
     * HashMap线程不安全的
     * 多线程对sessionTable读出或存入,可能产生异常
     * TcpSessionCache是在多线程环境下运行
     *
     * 2023-12-19实测,发现Hashtable并不线程安全,所以应用了HashMap和synchronized
     */
    private static HashMap<String, TcpSession> map = new HashMap<String, TcpSession>() ;
 
 
    /**
     * 得到信息
     * @return
     */
    public static Integer[] info(){
        Integer rtuTotalConnect = 0 ;//已经连接过中间件的RTU总数(包括在线与离线的)
        Integer rtuTotalOnLine = 0 ;//在线RTU总数
        Integer rtuTotalOffLine = 0 ;//离线RTU总数
        synchronized (map){
            rtuTotalConnect = map.size() ;
            Collection<TcpSession> col = map.values() ;
            for(TcpSession se : col){
                if(se.ioSession.isConnected()){
                    rtuTotalOnLine ++ ;
                }else{
                    rtuTotalOffLine ++ ;
                }
            }
        }
        return new Integer[] {rtuTotalConnect, rtuTotalOnLine, rtuTotalOffLine} ;
    }
 
    /**
     * 关闭所有网络连接
     */
    public static void closeAllSessions(){
        synchronized (map){
            Collection<TcpSession> col = map.values() ;
            for(TcpSession se : col){
                se.ioSession.closeNow() ;
            }
            map.clear();
        }
    }
 
    /**
     * 加入新的IoSession
     * @param rtuAddr
     * @param ioSession
     */
    //public static void putNewTcpSession(String rtuAddr, String protocolName, Short protocolVersion, IoSession ioSession){
    public static void putNewTcpSession(String rtuAddr, IoSession ioSession){
        synchronized (map){
            TcpSession tcpSe = map.get(rtuAddr) ;
            if(tcpSe == null){
                tcpSe = new TcpSession() ;
                //tcpSe.protocolName = protocolName ;
                //tcpSe.protocolVersion = protocolVersion ;
                tcpSe.ioSession = ioSession ;
                map.put(rtuAddr, tcpSe) ;
            }else{
                tcpSe.ioSession = ioSession ;
            }
        }
    }
    
    /**
     * 更新IoSession对应的rtuAddr
     * @param oldRtuAddr
     * @param newRtuAddr
     * @param ioSession
     */
    //public static void changeRtuAddr(String oldRtuAddr, String newRtuAddr, String protocolName, Short protocolVersion, IoSession ioSession){
    public static void changeRtuAddr(String oldRtuAddr, String newRtuAddr, IoSession ioSession){
        if(oldRtuAddr != null && newRtuAddr != null && !oldRtuAddr.equals(newRtuAddr)){
            synchronized (map){
                TcpSession tcpSe = map.get(oldRtuAddr) ;
                if(tcpSe == null){
                    putNewTcpSession(newRtuAddr, ioSession) ;
                }else{
                    map.remove(oldRtuAddr) ;
                    map.put(newRtuAddr, tcpSe) ;
                }
            }
        }
    }
    
    /**
     * 得到TcpSession
     * @param rtuAddr
     * @return
     */
    public static TcpSession getTcpSession(String rtuAddr){
        return map.get(rtuAddr) ;
    }
 
    /**
     * 得到Tcp通信协议名称
     * @param rtuAddr
     * @return
     * */
    public static Object[] getTcpProtocolNameVersion(String rtuAddr){
        TcpSession tcpSe = map.get(rtuAddr) ;
        if(tcpSe != null){
            return new Object[]{
                    tcpSe.ioSession.getAttribute(TcpIoSessionAttrIdIsRtuAddr.sessionArrProtocolName),
                    tcpSe.ioSession.getAttribute(TcpIoSessionAttrIdIsRtuAddr.sessionArrProtocolVersion)};
        }else{
            return null ;
        }
    }
 
 
    /**
     * 得到所有在线与离线数量统计
     * @return [0]=在线数量,[1]上线过,但当前离线的数量
     */
    public static Integer[] allOnLineStateStatistics(){
        synchronized (map){
            Integer[] arr = new Integer[]{0, 0} ;
            Iterator<Entry<String, TcpSession>> it = map.entrySet().iterator() ;
            Entry<String, TcpSession> entry = null ;
            while(it.hasNext()){
                entry = it.next() ;
                if(entry.getValue().ioSession.isConnected()){
                    arr[0]++ ;
                }else{
                    arr[1]++ ;
                }
            }
            return arr ;
        }
    }
    
    /**
     * 得到所有在线情况
     * @return
     */
    public static HashMap<String, Boolean> allOnLine(){
        synchronized (map){
            HashMap<String, Boolean> map = new HashMap<String, Boolean>();
            Iterator<Entry<String, TcpSession>> it = TcpSessionCache.map.entrySet().iterator() ;
            Entry<String, TcpSession> entry = null ;
            while(it.hasNext()){
                entry = it.next() ;
                map.put(entry.getKey(), entry.getValue().ioSession.isConnected()) ;
            }
            return map ;
        }
    }
 
 
    /**
     * 得到部分在线情况
     * @return
     */
    public static HashMap<String, Boolean> partOnLine(String[] rtuAddrArrGrp){
        synchronized (map){
            HashMap<String, Boolean> map = new HashMap<String, Boolean>();
            for(String rtuAddr : rtuAddrArrGrp){
                TcpSession tcpSe = TcpSessionCache.map.get(rtuAddr) ;
                if(tcpSe != null){
                    map.put(rtuAddr, tcpSe.ioSession.isConnected()) ;
                }
            }
            return map ;
        }
    }
 
    /**
     * 网络是否连接
     * @param rtuAddr
     * @return
     */
    public static Boolean isConnect(String rtuAddr){
        TcpSession tcpSe = map.get(rtuAddr) ;
        if(tcpSe != null){
            return tcpSe.ioSession.isConnected() ;
        }
        return null ;
    }
    
    /**
     * 通过IoSession输出数据
     * @param rtuAddr
     * @param data
     * @throws Exception
     */
    public static void write(String rtuAddr, byte[] data) throws Exception{
        TcpSession tcpSe = map.get(rtuAddr) ;
        if(tcpSe != null){
            if(tcpSe.ioSession.isConnected()){
                tcpSe.ioSession.write(data) ;
            }else{
                throw new Exception("Rtu连接已经关闭!") ;
            }
        }else{
            throw new Exception("Rtu未曾上线!") ;
        }
    }
    
    /**
     * 当有上行数据时
     * @param rtuAddr
     */
    public static void whenUpData(String rtuAddr){
        TcpSession tcpSe = map.get(rtuAddr) ;
        if(tcpSe != null){
            tcpSe.lastDownComTime = 0L ;//置0,使等待lastUpDataTimeLive(见config.xml配置文件)时长后,即刻下发缓存中的命令
            tcpSe.lastUpDataTime = System.currentTimeMillis() ;
            tcpSe.lastUpDataTimeForOnlineCtrl = System.currentTimeMillis() ;
        }
    }
    
    /**
     * 更新上行数据时刻
     * 当上行数据时刻已经过去一定时长,上行数据时刻清空
     * 当一定时间内没有上行数据,则认为RTU离线
     */
    public static void updateRtuStatus(Long now){
        synchronized (map){
            Set<Map.Entry<String, TcpSession>> entrySet = map.entrySet() ;
            Iterator<Map.Entry<String, TcpSession>> it = entrySet.iterator() ;
            Map.Entry<String, TcpSession> entry ;
            TcpSession tcpSe ;
            while(it.hasNext()){
                entry = it.next() ;
                tcpSe = entry.getValue();
                if(tcpSe.lastUpDataTime != null){
                    if(now - tcpSe.lastUpDataTime > ServerProperties.lastUpDataTimeLive){
                        tcpSe.lastUpDataTime = null ;
                    }
                }
                if(tcpSe.lastUpDataTimeForOnlineCtrl != null){
                    if(tcpSe.ioSession != null && tcpSe.ioSession.isConnected()){
                        if(now - tcpSe.lastUpDataTimeForOnlineCtrl > ServerProperties.disconnectedByNoUpDataMinutes){
                            tcpSe.ioSession.closeNow() ;
                            RtuLogDealer.log(entry.getKey(), "因较长时间未收上行数据,认为设备离线");
                        }
                    }
                }
            }
        }
    }
 
}