資訊專(zhuān)欄INFORMATION COLUMN
摘要：那沒(méi)有建立連接的情況下，發(fā)現(xiàn)房間這個(gè)功能是怎么實(shí)現(xiàn)的呢首先，既然手機(jī)處于局域網(wǎng)中，那么根據(jù)手機(jī)當(dāng)前在局域網(wǎng)的地址和子網(wǎng)掩碼，就可以獲得這個(gè)局域網(wǎng)內(nèi)所有機(jī)器的地址的范圍。
記得以前我們使用類(lèi)似“快牙”這些文件分享工具的時(shí)候，一開(kāi)始就是先在 手機(jī)A 上創(chuàng)建一個(gè)“房間”，然后連接上 手機(jī)A WiFi 熱點(diǎn)的其他手機(jī)（即這些手機(jī)處于一個(gè)局域網(wǎng)內(nèi)）就可以發(fā)現(xiàn)到這個(gè)房間并加入到這個(gè)房間里面，然后就可以互相分享文件了。那沒(méi)有建立連接的情況下，“發(fā)現(xiàn)房間”這個(gè)功能是怎么實(shí)現(xiàn)的呢？
首先，既然 手機(jī)A 處于局域網(wǎng)中，那么根據(jù) 手機(jī)A 當(dāng)前在局域網(wǎng)的 IP 地址和子網(wǎng)掩碼，就可以獲得這個(gè)局域網(wǎng)內(nèi)所有機(jī)器的 IP 地址 的范圍。如果在沒(méi)有建立連接的情況下，手機(jī)A 就可以給這個(gè)范圍內(nèi)的每個(gè) IP 地址都發(fā)送一個(gè)消息 —— 那么如果某個(gè) IP 地址的機(jī)器（設(shè)為 手機(jī)B）會(huì)對(duì)這個(gè)消息做出回應(yīng)，便說(shuō)明 手機(jī)B 是 手機(jī)A 的“自己人”，那么 手機(jī)A 便可以告訴 手機(jī)B 它在當(dāng)前的局域網(wǎng)建了一個(gè)“房間”，房間號(hào)是個(gè)啥，然后 手機(jī)B 可以選擇是否加入到這個(gè)“房間”。
在Java網(wǎng)絡(luò)編程（1）中，我們已經(jīng)知道可以使用 NetworkInterface 來(lái)獲得機(jī)器在局域網(wǎng)內(nèi) IP 地址；
在Java網(wǎng)絡(luò)編程（2）中，我們知道使用 UDP，便可以在不建立連接的情況下，直接向某個(gè) IP 地址發(fā)送消息；
如果每次都是遍歷這個(gè)局域網(wǎng)內(nèi)所有的 IP 地址，并使用 UDP 向每個(gè) IP 發(fā)送消息，那樣就有點(diǎn)麻煩了。事實(shí)上，我們可以使用廣播。每個(gè)局域網(wǎng)都有一個(gè)對(duì)應(yīng)的廣播地址，向廣播地址發(fā)送的數(shù)據(jù)包通過(guò)網(wǎng)關(guān)設(shè)備（比如路由器）時(shí)，網(wǎng)關(guān)設(shè)備會(huì)向局域網(wǎng)的每臺(tái)設(shè)備發(fā)送一份該數(shù)據(jù)包的副本。通過(guò) IP 和子網(wǎng)掩碼計(jì)算廣播地址的方法簡(jiǎn)單的形容就是 （IP地址）|（~子網(wǎng)掩碼）—— 將子網(wǎng)掩碼按位取反再和IP地址進(jìn)行或運(yùn)算，比如當(dāng)前機(jī)器在局域網(wǎng)內(nèi)的地址為 192.168.1.3，子網(wǎng)掩碼為 255.255.255.0（取反后為 0.0.0.255），那么廣播地址為 192.168.1.255。廣播也是在不建立連接的情況下就發(fā)送數(shù)據(jù)，所以廣播不能通過(guò) TCP 實(shí)現(xiàn)，只能是 UDP。在 Java 中，通過(guò) UDP 進(jìn)行廣播和單播（即只向一個(gè) IP 地址發(fā)送數(shù)據(jù)包）的程序幾乎沒(méi)有區(qū)別，只是地址由一個(gè)特定的單播地址（如 192.168.1.3)變?yōu)榱似鋵?duì)應(yīng)的廣播地址（192.168.1.255）。
現(xiàn)在讓我們來(lái)實(shí)現(xiàn)下面的功能：
1、Broadcaster 創(chuàng)建一個(gè)房間，并每隔 1 秒向局域網(wǎng)廣播一個(gè)特定的消息；
2、同一個(gè)局域網(wǎng)的 Device 如果收到了 3 次這個(gè)特定的消息，之后便向 Broadcaster 發(fā)送加入房間的消息；
3、Broadcaster 收到 Device 請(qǐng)求加入房間的消息后，將 Device 加入房間。
首先定義發(fā)送者類(lèi)和接收者類(lèi)，他們都實(shí)現(xiàn)了 Runnable，分別可以用來(lái)發(fā)送和接收：
Sender.java
import java.io.IOException;
import java.net.*;
public class Sender implements Runnable {
private static final byte[] EMPTY_DATA = new byte[0];
private final DatagramSocket socket;
private final SocketAddress broadcastAddress;
private final long sendingInterval; // unit is ms
public Sender(DatagramSocket socket,
SocketAddress broadcastAddress, int sendingInterval) {
this.socket = socket;
this.broadcastAddress = broadcastAddress;
this.sendingInterval = sendingInterval;
}
@Override
public void run() {
while (true) {
byte[] data = getNextData();
if (data == null || data.length == 0) {
break;
}
DatagramPacket outPacket = new DatagramPacket(
data, data.length, broadcastAddress);
try {
socket.send(outPacket);
System.out.println("Sender: Data has been sent");
Thread.sleep(sendingInterval);
} catch (IOException | InterruptedException ex) {
System.err.println("Sender: Error occurred while sending packet");
break;
}
}
System.out.println("Sender: Thread is end");
}
/**
* 獲得下一次發(fā)送的數(shù)據(jù)
* 子類(lèi)需要重寫(xiě)這個(gè)方法，返回下一次要發(fā)送的數(shù)據(jù)
*
* @return 下一次發(fā)送的數(shù)據(jù)
*/
public byte[] getNextData() {
return EMPTY_DATA;
}
}
Receiver.java
import java.io.IOException;
import java.net.*;
public class Receiver implements Runnable {
private final int BUF_SIZE = 512;
private final DatagramSocket socket;
public Receiver(DatagramSocket socket) {
this.socket = socket;
}
@Override
public void run() {
byte[] inData = new byte[BUF_SIZE];
DatagramPacket inPacket = new DatagramPacket(inData, inData.length);
while (true) {
try {
socket.receive(inPacket);
if (!handlePacket(inPacket)) {
break;
}
} catch (IOException ex) {
System.out.println("Receiver: Socket was closed.");
break;
}
}
System.out.println("Receiver: Thread is end");
}
/**
* 處理接收到的數(shù)據(jù)報(bào)
* 子類(lèi)需要重寫(xiě)這個(gè)方法，處理接收到的數(shù)據(jù)包，并返回是否繼續(xù)接收
*
* @param packet 接收到的數(shù)據(jù)報(bào)
* @return 是否需要繼續(xù)接收
*/
public boolean handlePacket(DatagramPacket packet) {
return false;
}
}
然后我們定義 Device 和 Broadcaster：
Device.java
import java.io.IOException;
import java.net.*;
public class Device {
private static final int DEFAULT_LISTENING_PORT = 10000;
private final InetAddress address;
private final int port;
private DatagramSocket socket;
public Device(int port) throws IOException {
this.port = port;
this.address = InetAddress.getLocalHost();
}
public Device(InetAddress address, int port) {
this.address = address;
this.port = port;
}
public void start() throws SocketException, InterruptedException {
System.out.println("Device has been started...");
InetAddress lanAddr = LANAddressTool.getLANAddressOnWindows();
if (lanAddr != null) {
System.out.println("Device: LAN Address: " + lanAddr.getHostAddress());
}
socket = new DatagramSocket(port);
Receiver receiver = new Receiver(socket) {
int recvCount = 0;
@Override
public boolean handlePacket(DatagramPacket packet) {
String recvMsg = new String(packet.getData(), 0, packet.getLength());
if ("ROOM".equals(recvMsg)) {
System.out.printf("Device: Received msg "%s"
", recvMsg);
recvCount++;
if (recvCount == 3) {
byte[] data = "JOIN".getBytes();
DatagramPacket respMsg = new DatagramPacket(
data, data.length, packet.getSocketAddress()); // 此時(shí) packet 包含了發(fā)送者地址和監(jiān)聽(tīng)端口
try {
socket.send(respMsg);
System.out.println("Device: Sent response "JOIN"");
} catch (IOException ex) {
ex.printStackTrace(System.err);
}
return false; // 停止接收
}
}
return true;
}
};
Thread deviceThread = new Thread(receiver);
deviceThread.start(); // 啟動(dòng)接收數(shù)據(jù)包的線(xiàn)程
deviceThread.join();
close();
System.out.println("Device has been closed.");
}
public void close() {
if (socket != null) {
socket.close();
}
}
@Override
public String toString() {
return "Device {" + "address=" + address + ", port=" + port + "}";
}
public static void main(String[] args) throws Exception {
Device device = new Device(DEFAULT_LISTENING_PORT);
device.start();
}
}
Broadcaster.java
import java.net.*;
public class Broadcaster {
private static final int DEFAULT_BROADCAST_PORT = 10000;
private final InetAddress bcAddr;
private final int bcPort;
private DatagramSocket socket;
public Broadcaster(InetAddress broadcastAddress, int broadcastPort) {
this.bcAddr = broadcastAddress;
this.bcPort = broadcastPort;
}
public void start() throws SocketException, InterruptedException {
System.out.println("Broadcaster has been started...");
final Room room = new Room("Test");
System.out.printf("Broadcaster: Created room "%s"
", room.getName());
socket = new DatagramSocket();
SocketAddress bcSocketAddr = new InetSocketAddress(bcAddr, bcPort);
Sender sender = new Sender(socket, bcSocketAddr, 1000) {// 每隔 1000ms 廣播一次
final byte[] DATA = "ROOM".getBytes();
@Override
public byte[] getNextData() {
return DATA;
}
};
Receiver recver = new Receiver(socket) {
@Override
public boolean handlePacket(DatagramPacket packet) {
String recvMsg = new String(packet.getData(), 0, packet.getLength());
if ("JOIN".equals(recvMsg)) {
Device device = new Device(packet.getAddress(), packet.getPort());
room.addDevice(device);
room.listDevices();
}
return true; // 一直接收
}
};
Thread senderThread = new Thread(sender);
Thread recverThread = new Thread(recver);
senderThread.start(); // 啟動(dòng)發(fā)送（廣播）數(shù)據(jù)包的線(xiàn)程
recverThread.start(); // 啟動(dòng)接收數(shù)據(jù)包的線(xiàn)程
senderThread.join();
recverThread.join();
close();
}
public void close() {
if (socket != null) {
socket.close();
}
}
public static void main(String[] args) throws Exception {
InetAddress bcAddr = LANAddressTool.getLANBroadcastAddressOnWindows();
if (bcAddr != null) {
System.out.println("Broadcast Address: " + bcAddr.getHostAddress());
Broadcaster broadcaster = new Broadcaster(bcAddr, DEFAULT_BROADCAST_PORT);
broadcaster.start();
} else {
System.out.println("Please check your LAN~");
}
}
}
Room.java
import java.util.*;
public class Room {
private final String name;
private final List devices;
public Room(String name) {
this.name = name;
this.devices = new ArrayList<>();
}
public boolean addDevice(Device device) {
return devices.add(device);
}
public String getName() {
return name;
}
public void listDevices() {
System.out.printf("Room (%s), current devices:
", name);
for (Device device : devices) {
System.out.println(device);
}
}
}
（完整的 Demo 可以訪問(wèn)：https://github.com/mizhoux/LA...）
我們將這個(gè) Demo 打包成 jar，然后開(kāi)始運(yùn)行：
1、首先我們?cè)诒緳C(jī)上啟動(dòng) Broadcaster：
2、我們將本機(jī)作為一個(gè) Device 啟動(dòng)：
可以看到此時(shí) Broadcaster 創(chuàng)建的房間已經(jīng)有了一個(gè) Device：
3、我們啟動(dòng)局域網(wǎng)內(nèi)的另外一臺(tái)設(shè)備：
此時(shí) Broadcaster 創(chuàng)建的房間便有兩個(gè) Device：
4、再啟動(dòng)局域網(wǎng)內(nèi)的一臺(tái)設(shè)備：
此時(shí)房間里則有三個(gè) Device：
因?yàn)?UDP 在不需要建立連接的基礎(chǔ)上就可以發(fā)送消息，所以它可以方便的用來(lái)探測(cè)局域網(wǎng)內(nèi)特定類(lèi)型的機(jī)器 —— 這是個(gè)很有用的功能 —— 又比如一個(gè)集群當(dāng)中可能會(huì)突然有機(jī)器宕機(jī)，為了檢測(cè)這一事件的發(fā)生，就需要集群 master機(jī)器 每隔一定的時(shí)間向每臺(tái)機(jī)器發(fā)送若干心跳檢測(cè)包，如果有回復(fù)說(shuō)明機(jī)器正常，否則說(shuō)明該機(jī)器出現(xiàn)了故障，此時(shí)不需要連接而且高效的 UDP 就十分適合這種場(chǎng)合。當(dāng)然，我們始終還是要考慮到 UDP 是不可靠的協(xié)議，它并不能代替 TCP —— 永遠(yuǎn)需要根據(jù)環(huán)境，來(lái)選擇最合適的技術(shù)。
文章版權(quán)歸作者所有，未經(jīng)允許請(qǐng)勿轉(zhuǎn)載,若此文章存在違規(guī)行為，您可以聯(lián)系管理員刪除。
轉(zhuǎn)載請(qǐng)注明本文地址：http://specialneedsforspecialkids.com/yun/66340.html
摘要：現(xiàn)在在本機(jī)同一局域網(wǎng)的一臺(tái)機(jī)器和阿里云主機(jī)上都運(yùn)行然后啟動(dòng)發(fā)送端接收端接收結(jié)果可以看到每個(gè)接收端都正確的接收了發(fā)送端發(fā)送的消息。 今天的主角是 UDP（User Datagram Protocol，用戶(hù)數(shù)據(jù)報(bào)協(xié)議）。我們都知道 TCP 是一種可靠的協(xié)議 —— 首先客戶(hù)端和服務(wù)端需要建立連接（三次握手），數(shù)據(jù)發(fā)送完畢需要斷開(kāi)連接（四次揮手）；如果發(fā)送數(shù)據(jù)時(shí)數(shù)據(jù)損壞或者丟失，那么 TCP ...
閱讀 1408·2023-04-26 01:58
閱讀 2282·2021-11-04 16:04
閱讀 1753·2021-08-31 09:42
閱讀 1765·2021-07-25 21:37
閱讀 1066·2019-08-30 15:54
閱讀 2074·2019-08-30 15:53
閱讀 3047·2019-08-29 13:28
閱讀 2687·2019-08-29 10:56
