用java的Socket编写的小小小聊天工具，在本机调试成功，但是在两台能互相ping通的电脑却不能连接上
Posted
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了用java的Socket编写的小小小聊天工具，在本机调试成功，但是在两台能互相ping通的电脑却不能连接上相关的知识，希望对你有一定的参考价值。
即实现聊天【一台做服务器，一台做客户端】），防火墙也关闭了的，还是不行！求高手赐教！！！
建议先用一个最简单的程序测试客户端链接服务器，这样可以排除很多干扰，下面两个类已经是客户端和服务器的最简代码了，你先测一下，如果仍不能连接，把异常信息发上来看看。另：Socket连接容易出问题的地方：
1）端口号冲突，服务器无法启动；
2）IP地址错误，IP地址分本机地址、局域网地址和广域网地址，在不同的环境下使用不同的地址测试，如果错误将无法被客户端访问，还有如果机器安装了多网卡等问题都会引起错误；
3）防火墙禁止Java程序开放端口，不单是服务器机器的软防火墙，局域网的防火墙也可能阻止外部的访问。
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
public class ServerMain
public ServerMain()
try
ServerSocket ss = new ServerSocket(31647);
Socket s = ss.accept();
System.out.printf("客户端连接成功：\n%s\n", s.toString());
catch (IOException e)
e.printStackTrace();
public static void main(String[] args)
new ServerMain();
import java.io.IOException;
import java.net.Socket;
import java.net.UnknownHostException;
public class ClientMain
public ClientMain()
try
Socket s = new Socket("192.168.1.100", 31647); // 修改IP地址
System.out.println("连接服务器成功！");
catch (UnknownHostException e)
e.printStackTrace();
catch (IOException e)
e.printStackTrace();
public static void main(String[] args)
new ClientMain();
参考技术A 是不是IP地址没做相应修改？
python socket 小小实验
学python这么久，也没有去尝试过socket，今天就来做个小工具吧。
假如说你在局域网内有20台以上的linux机器，你要快速知道这么多机器的磁盘使用率有没有超过 某一个数值，进程是不是挂了，反正就是信息收集。
#######################################
###################### 主控端
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import json
from socket import *
HOST = 'localhost' #鏈嶅姟鍣ㄧ殑涓绘満鍚�
PORT = 21571 #绔彛鍙�
BUFSIZ = 1024 #缂撳啿鍖�
ADDR = (HOST,PORT) #鍦板潃址
def zabbix_agent_master_tcp():
HOST = 'localhost' #鏈嶅姟鍣ㄧ殑涓绘満鍚�
PORT = 21571 #绔彛鍙�
BUFSIZ = 1024 #缂撳啿鍖�
ADDR = (HOST,PORT) #鍦板潃
tcpCliSocket = socket(AF_INET,SOCK_STREAM) #鍒涘缓瀹㈡埛绔鎺ュ瓧
tcpCliSocket.connect(ADDR) #杩炴帴鏈嶅姟鍣�
tcpCliSocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
while True: #閫氫俊寰幆
data = input('> ') #瀹㈡埛绔緭鍏ヤ俊鎭�
if data is None or data =='':
data ='11'
data = str.encode(data)
tcpCliSocket.send(data) #鍙戦�佸鎴风淇℃伅
data = tcpCliSocket.recv(BUFSIZ) #鎺ュ彈鏈嶅姟鍣ㄨ繑鍥炰俊鎭�
print('get:',data.decode('utf-8'))
if data=='quit': #濡傛灉杈撳叆淇℃伅涓虹┖锛屽垯璺冲嚭寰幆锛屽叧闂�氫俊
break
tcpCliSocket.close()
def diskusecheck_request( dict_request = {
'type':'diskusecheck',
'diskname':'root',
'limit':'20',
'result':'alarm',
'operation':'request',
'ip':HOST}):
return json.dumps(dict_request)
def processcheck_request( dict_request = {
'type':'processcheck',
'processname':'staf',
'limit':'90',
'result':'alarm',
'operation':'request',
'ip':HOST}):
return json.dumps(dict_request)
def zabbix_master_udp():
import socket
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
network = '<broadcast>'
#buffer1 = "'type:'diskusecheck','diskname':'root','limit':'20','result':'0','operation':'request', 'master':'192.168.0.78',"
dict_request = {'type':'diskusecheck',
'diskname':'root',
'limit':'20',
'result':'alarm',
'operation':'request',
'ip':HOST,}
dict_request['diskname'] = 'root'
buffer1=json.dumps(dict_request)
dict_request['diskname'] = 'usr1'
buffer2=json.dumps(dict_request )
dict_request['type'] = 'diskusecheck'
dict_request['diskname'] = 'staf'
buffer3=json.dumps(dict_request)
##########################################################################
while True:
data = str(raw_input(' '))
if data is None or data =='':
data ='11'
s.sendto(data.encode('utf-8'), (network, PORT))
data = s.recv(4096)
print data
if data=='root':
s.sendto(buffer1.encode('utf-8'), (network, PORT))
data = s.recv(4096)
print data
if data=='usr1':
s.sendto(buffer2.encode('utf-8'), (network, PORT))
data = s.recv(4096)
print data
if data=='staf':
s.sendto(buffer3.encode('utf-8'), (network, PORT))
data = s.recv(4096)
print data
##########################################################################
# #######################################################################
s.sendto(buffer1.encode('utf-8'), (network, PORT))
data = s.recv(4096)
tmp_data =json.loads(data.decode('utf-8'))
if tmp_data['res']=='alarm' and tmp_data['type']=='diskusecheck':
pass
elif tmp_data['res']=='die' and tmp_data['type']=='diskusecheck':
pass
# #######################################################################
s.sendto(buffer2.encode('utf-8'), (network, PORT))
data = s.recv(4096)
tmp_data =json.loads(data.decode('utf-8'))
if tmp_data['res']=='alarm' and tmp_data['type']=='diskusecheck':
pass
elif tmp_data['res']=='die' and tmp_data['type']=='diskusecheck':
pass
# #######################################################################
s.sendto(buffer3.encode('utf-8'), (network, PORT))
data = s.recv(4096)
tmp_data =json.loads(data.decode('utf-8'))
if tmp_data['res']=='alarm' and tmp_data['type']=='diskusecheck':
pass
elif tmp_data['res']=='die' and tmp_data['type']=='diskusecheck':
pass
s.close()
zabbix_master_udp()
# a ={'a':'6'}
# print type(json.dumps(a))
#######################################################################%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
Agent keko
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import json
import re, os
from socket import *
if os.name=='nt':
HOST='localhost'
else:
HOST = re.findall(r'inet.*?(\d+.\d+.\d+.\d+)',os.popen('ifocnfig').read())[0] #服务器的主机名
PORT = 21571 #端口号
BUFSIZ = 1024 #缓冲区
def zabbix_agent_tcp():
PORT = 21571 #端口号
BUFSIZ = 2048 #缓冲区大小，1kb
tcpSerSocket = socket(AF_INET,SOCK_STREAM) #创建 tcp 套接字
tcpSerSocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
tcpSerSocket.listen(5) #设置并启动套接字监听
while True: #无限循环，等待客户端连接
print('waiting for connection...')
tcpCliSocket,addr = tcpSerSocket.accept() #被动接受客户端连接
print('...connected from:',addr)
while True: #对话循环，等待客户端发送消息
data = tcpCliSocket.recv(BUFSIZ) #接收客户端消息
print data
if data=='ifconfig':
tcpCliSocket.send("'caozuo':'ifconfig'") #如果收到消息，将消息原封不动返回客户端
elif data=='':
tcpCliSocket.send("'caozuo':'what are you say'")
elif data=='exit':
tcpCliSocket.close()
sys.exit()
else:
tcpCliSocket.send("'caozuo':'invalaid'")
def diskusecheck_response( dict_request = {
'type':'diskusecheck',
'diskname':'root',
'limit':'20',
'result':None,
'operation':'response',
'ip':HOST}):
keyword = dict_request['diskname']
if dict_request['diskname']=='root':
keyword ='/'
elif dict_request['diskname']=='usr1':
keyword ='/usr1'
res_alarm = os.popen("df %s |grep -v grep |grep -v Used |awk '{print $5}' |sed 's/\'%\'//g' ")
if int(res_alarm)>=int(dict_request['limit']):
dict_request['result']='alarm'
return json.loads(dict_request)
def processcheck_response( dict_request = {
'type':'processcheck',
'processname':'staf',
'limit':'90',
'result':'alarm',
'operation':'request',
'ip':HOST}):
res_alarm = os.popen("ps -ef |grep %s |grep -v grep |wc -l' ")
if int(res_alarm)==0:
dict_request['result']='die'
else:
dict_request['result']='alive'
return json.loads(dict_request)
def zabbix_agent_udp():
import socket
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind(('', PORT))
print('Listening for broadcast at ', s.getsockname())
while True:
data, address = s.recvfrom(65535)
print 'Server received from {}:{}'.format(address, data.decode('utf-8'))
request_data = json.loads(data.encode('utf-8'))
if os.name=='nt':
data_win='your pc is win,please run @linux'
s.sendto(data_win.encode('utf-8'), (address))
print 'Server sent to {}:{}'.format(address, data_win.decode('utf-8'))
continue
if request_data['type']=='diskusecheck':
data_response =diskusecheck_response( dict_request = {
'type':'diskusecheck',
'diskname':request_data['diskname'],
'limit':request_data['limit'],
'result':None,
'operation':'response',
'ip':HOST})
s.sendto(data_response.encode('utf-8'), (address))
print('Server sent to {}:{}'.format(address, data_response.decode('utf-8')).replace('request','response'))
zabbix_agent_udp()
以上是关于用java的Socket编写的小小小聊天工具，在本机调试成功，但是在两台能互相ping通的电脑却不能连接上的主要内容，如果未能解决你的问题，请参考以下文章