我正在用 Java 编写 Bully 算法程序
这是代码:
package newbully;
public class NewBully {
public static void main(String[] args) {
int total_processes = 6;
RunningThread[] t = new RunningThread[total_processes];
for (int i = 0; i < total_processes; i++) {
t[i] = new RunningThread(new Process(i+1, i+1), total_processes);//passing process id, priority, total no. of processes to running thread
}
try {
Election.initialElection(t);
} catch (Exception e) {
System.out.println("Possibly you are using null references in array");
}
for (int i = 0; i < total_processes; i++) {
new Thread(t[i]).start();//start every thread
}
}
}
package newbully;
import java.util.*;
import java.io.*;
import java.net.*;
public class RunningThread implements Runnable {
private Process process;
private int total_processes;
ServerSocket[] sock;
Random r;
public Process getProcess() {
return process;
}
public void setProcess(Process process) {
this.process = process;
}
public RunningThread(Process process, int total_processes) {
this.process = process;
this.total_processes = total_processes;
this.r = new Random();
this.sock = new ServerSocket[total_processes];
}
synchronized private void recovery() {
System.out.println("Process[" + this.process.getPid() + "]: -> Recovered from Crash");
//Find current co-ordinator.
}
synchronized private void pingCoOrdinator() {
try {
if (Election.isPingFlag()) {
synchronized (Election.lock) {
Election.lock.wait();
}
}
if (!Election.isElectionFlag()) {
Election.setPingFlag(true);
System.out.println("Process[" + this.process.getPid() + "]: Are you alive?");
Socket outgoing = new Socket(InetAddress.getLocalHost(), 12345);
outgoing.close();
Election.setPingFlag(false);
synchronized (Election.lock) {
Election.lock.notifyAll();
}
}
} catch (Exception ex) {
//Initiate Election
System.out.println("process[" + this.process.getPid() + "]: -> Co-Ordinator is down\nInitiating Election");
Election.setElectionFlag(true);
Election.setPingFlag(false);
synchronized (Election.lock) {
Election.lock.notifyAll();
}
}
}
synchronized private void executeJob() {
int temp = r.nextInt(20);
for (int i = 0; i <= temp; i++) {
try {
Thread.sleep(700);
} catch (InterruptedException e) {
System.out.println("Error Executing Thread:" + process.getPid());
System.out.println(e.getMessage());
}
}
}
synchronized private boolean sendMessage() {
boolean response = false;
int i = 0;
try {
if (Election.isMessageFlag()) {
synchronized (Election.lock) {
Election.lock.wait();
}
}
Election.setMessageFlag(true);
if (Election.isElectionFlag()) {
for (i = this.process.getPid() + 1; i <= this.total_processes; i++) {
try {
Socket electionMessage = new Socket(InetAddress.getLocalHost(), 10000 + i);
System.out.println("Process[" + this.process.getPid() + "] -> Process[" + i + "] responded to election message successfully");
electionMessage.close();
response = true;
} catch (Exception ex) {
System.out.println("Process[" + this.process.getPid() + "] -> Process[" + i + "] did not respond to election message");
}
}
}
Election.setMessageFlag(false);
synchronized (Election.lock) {
Election.lock.notifyAll();
}
} catch (Exception ex1) {
System.out.println(ex1.getMessage());
}
return response;
}
synchronized private void serve() {
try {
//service counter
Socket incoming = null;
ServerSocket s = new ServerSocket(12345);
for (int counter = 0; counter < 10; counter++) {
incoming = s.accept();
System.out.println("Process[" + this.process.getPid() + "]:Yes");
Scanner scan = new Scanner(incoming.getInputStream());
PrintWriter out = new PrintWriter(incoming.getOutputStream(), true);
if (scan.hasNextLine()) {
if (scan.nextLine().equals("Who is the co-ordinator?")) {
System.out.print("Process[" + this.process.getPid() + "]:");
out.println(this.process);
}
}
}
//after serving 10 requests go down for random time
this.process.setCoOrdinatorFlag(false);
this.process.setDownflag(true);
try {
incoming.close();
s.close();
sock[this.process.getPid() - 1].close();
Thread.sleep((this.r.nextInt(10) + 1) * 1000000);//going down
recovery();
} catch (InterruptedException e) {
System.out.println(e.getMessage());
}
} catch (IOException ex) {
System.out.println(ex.getMessage());
}
}
@Override
public void run() {
try {
sock[this.process.getPid() - 1] = new ServerSocket(10000 + this.process.getPid());
} catch (IOException ex) {
System.out.println(ex.getMessage());
}
while (true) {
if (process.isCoOrdinatorFlag()) {
//serve other processes
serve();
} else {
while (true) {
//Execute some task
executeJob();
//Ping the co-ordinator
pingCoOrdinator();
if (Election.isElectionFlag()) {
if (!sendMessage()) {//elect self as co-ordinator
System.out.println("New Co-Ordinator: Process[" + this.process.getPid() + "]");
this.process.setCoOrdinatorFlag(true);
Election.setElectionFlag(false);
break;
}
}
}
}
}
}
}
package newbully;
public class Election {
private static boolean pingFlag = false;
private static boolean electionFlag = false;
private static boolean messageFlag = false;
public static final Object lock = new Object();
public static boolean isMessageFlag() {
return messageFlag;
}
public static void setMessageFlag(boolean messageFlag) {
Election.messageFlag = messageFlag;
}
public static boolean isPingFlag() {
return pingFlag;
}
public static void setPingFlag(boolean pingFlag) {
Election.pingFlag = pingFlag;
}
public static boolean isElectionFlag() {
return electionFlag;
}
public static void setElectionFlag(boolean electionFlag) {
Election.electionFlag = electionFlag;
}
public static void initialElection(RunningThread[] t) {
Process temp = new Process(-1, -1);
for (int i = 0; i < t.length; i++) {
if (temp.getPriority() < t[i].getProcess().getPriority()) {
temp = t[i].getProcess();
}
}
t[temp.pid - 1].getProcess().CoOrdinatorFlag = true;
}
}
package newbully;
public class Process {
int pid;
boolean downflag,CoOrdinatorFlag;
public boolean isCoOrdinatorFlag() {
return CoOrdinatorFlag;
}
public void setCoOrdinatorFlag(boolean isCoOrdinator) {
this.CoOrdinatorFlag = isCoOrdinator;
}
int priority;
public boolean isDownflag() {
return downflag;
}
public void setDownflag(boolean downflag) {
this.downflag = downflag;
}
public int getPid() {
return pid;
}
public void setPid(int pid) {
this.pid = pid;
}
public int getPriority() {
return priority;
}
public void setPriority(int priority) {
this.priority = priority;
}
public Process() {
}
public Process(int pid, int priority) {
this.pid = pid;
this.downflag = false;
this.priority = priority;
this.CoOrdinatorFlag = false;
}
}
这是输出:
//--When delay in executeJob() method is 100
Process[4]: Are you alive?
Process[6]:Yes
Process[4]: Are you alive?
Process[6]:Yes
Process[3]: Are you alive?
Process[6]:Yes
Process[5]: Are you alive?
Process[6]:Yes
Process[1]: Are you alive?
Process[6]:Yes
Process[4]: Are you alive?
Process[6]:Yes
Process[3]: Are you alive?
Process[6]:Yes
Process[3]: Are you alive?
Process[6]:Yes
Process[2]: Are you alive?
Process[6]:Yes
Process[5]: Are you alive?
Process[6]:Yes
Process[1]: Are you alive?
process[1]: -> Co-Ordinator is down
Initiating Election
Process[1] -> Process[2] responded to election message successfully
Process[1] -> Process[3] responded to election message successfully
Process[1] -> Process[4] responded to election message successfully
Process[1] -> Process[5] responded to election message successfully
Process[1] -> Process[6] did not respond to election message
Process[2] -> Process[3] responded to election message successfully
Process[3] -> Process[4] responded to election message successfully
Process[4] -> Process[5] responded to election message successfully
Process[2] -> Process[4] responded to election message successfully
Process[2] -> Process[5] responded to election message successfully
Process[3] -> Process[5] responded to election message successfully
Process[5] -> Process[6] did not respond to election message
New Co-Ordinator: Process[5]
New Co-Ordinator: Process[1]
Address already in use: JVM_Bind
Address already in use: JVM_Bind
Address already in use: JVM_Bind
//--When delay in executeJob() method is 700
Process[3]: Are you alive?
Process[6]:Yes
Process[5]: Are you alive?
Process[6]:Yes
Process[2]: Are you alive?
Process[1]: Are you alive?
Process[6]:Yes
Process[6]:Yes
Process[5]: Are you alive?
Process[1]: Are you alive?
Process[6]:Yes
Process[6]:Yes
Process[4]: Are you alive?
Process[6]:Yes
Process[3]: Are you alive?
Process[6]:Yes
Process[2]: Are you alive?
Process[6]:Yes
Process[1]: Are you alive?
Process[6]:Yes
Process[4]: Are you alive?
process[4]: -> Co-Ordinator is down
Initiating Election
Process[4] -> Process[5] responded to election message successfully
Process[4] -> Process[6] did not respond to election message
Process[5] -> Process[6] did not respond to election message
New Co-Ordinator: Process[5]
Process[1]: Are you alive?
Process[5]:Yes
Process[1]: Are you alive?
Process[5]:Yes
Process[3]: Are you alive?
Process[5]:Yes
Process[2]: Are you alive?
Process[5]:Yes
Process[1]: Are you alive?
Process[5]:Yes
Process[4]: Are you alive?
Process[5]:Yes
Process[2]: Are you alive?
Process[5]:Yes
Process[4]: Are you alive?
Process[5]:Yes
Process[3]: Are you alive?
Process[5]:Yes
Process[3]: Are you alive?
Process[5]:Yes
Process[2]: Are you alive?
process[2]: -> Co-Ordinator is down
Initiating Election
Process[2] -> Process[3] responded to election message successfully
Process[2] -> Process[4] responded to election message successfully
Process[2] -> Process[5] did not respond to election message
Process[2] -> Process[6] did not respond to election message
Process[3] -> Process[4] responded to election message successfully
Process[3] -> Process[5] did not respond to election message
Process[3] -> Process[6] did not respond to election message
Process[1] -> Process[2] responded to election message successfully
Process[1] -> Process[3] responded to election message successfully
Process[1] -> Process[4] responded to election message successfully
Process[1] -> Process[5] did not respond to election message
Process[1] -> Process[6] did not respond to election message
Process[2] -> Process[3] responded to election message successfully
Process[2] -> Process[4] responded to election message successfully
Process[2] -> Process[5] did not respond to election message
Process[2] -> Process[6] did not respond to election message
Process[4] -> Process[5] did not respond to election message
Process[4] -> Process[6] did not respond to election message
New Co-Ordinator: Process[4]
Process[3]: Are you alive?
Process[4]:Yes
Process[3]: Are you alive?
Process[4]:Yes
Process[1]: Are you alive?
Process[4]:Yes
Process[2]: Are you alive?
Process[4]:Yes
Process[1]: Are you alive?
Process[4]:Yes
Process[2]: Are you alive?
Process[4]:Yes
Process[2]: Are you alive?
Process[4]:Yes
Process[2]: Are you alive?
Process[4]:Yes
Process[3]: Are you alive?
Process[4]:Yes
Process[1]: Are you alive?
Process[4]:Yes
Process[3]: Are you alive?
process[3]: -> Co-Ordinator is down
Initiating Election
Process[3] -> Process[4] did not respond to election message
Process[3] -> Process[5] did not respond to election message
Process[3] -> Process[6] did not respond to election message
New Co-Ordinator: Process[3]
New Co-Ordinator: Process[2]
Address already in use: JVM_Bind
Address already in use: JVM_Bind
Address already in use: JVM_Bind
Address already in use: JVM_Bind
Address already in use: JVM_Bind
最后我开始收到地址已在使用中的异常:JVM_Bin。
另外，如果我们在抛出异常之前检查输出中最新选出的协调员，它会在询问协调员是否还活着？
之前选择两次
我确信当协调员死亡时，我已经为其提供了足够的延迟，以便它不会在两者之间醒来。
当我给予额外的交易时，程序就会继续进行，否则它会停止在中间。
那么为什么一定会出现这个问题呢？
我找到了异常的原因
之所以发生这种情况，是因为如果您仔细查看异常消息之前的输出，它会选择协调器 2ce。
每当一个线程被选为协调者时，它就会在端口 12345 处打开一个 ServerSocket。
由于它发生了 2ce，因此可能会抛出异常。
但我不明白...为什么它选了2ce？
最佳答案
两种情况下的错误消息都是“地址已在使用中:JVM_Bind”。
此消息通常是 java.net.BindException 的一部分，当您尝试创建/打开 Socket 并且端口已在使用中时会抛出该异常。在这种情况下，您可能会尝试打开或创建同一个套接字两次。
发生这种情况是因为在打开和关闭套接字之间，异常阻止了 socket.close() 的调用。在这种情况下，当您为选举领导者创建一个套接字时，但由于主机“崩溃”，会抛出异常，因此永远不会调用 close() 。
我认为你需要添加这一行
electionMessage.close();到发送消息这部分的catch子句。
try {
Socket electionMessage = new Socket(InetAddress.getLocalHost(), 10000 + i);
System.out.println("Process[" + this.process.getPid() + "] -> Process[" + i + "] responded to election message successfully");
electionMessage.close();
response = true;
} catch (Exception ex) {
//Add close here
System.out.println("Process[" + this.process.getPid() + "] -> Process[" + i + "] did not respond to election message");
}
此外，我建议将 close 添加到所有其他相关的 catch 子句中，以防在其他地方发生这种情况，并且始终是避免相关问题的良好做法。 我还建议明确指定您想要在每个地方捕获哪些异常，这样您就不会陷入其他陷阱。
我希望这是一个好的起点。
编辑回应第一条评论
我认为您“锁定”线程的方式存在问题。您依靠 boolean 标志来告诉您已到达代码中的特定点。但是 boolean 标志本身不受同步块(synchronized block)的控制，也不受代码上任何其他锁的控制。因此，多个线程可以传递锁并导致意外事件，例如多次尝试打开同一个套接字。
您正在使用带有对象的代码来锁定
if (Election.isMessageFlag()) {
synchronized (Election.lock) {
Election.lock.wait();
}
}
Election.setMessageFlag(true);
if (Election.isElectionFlag()) {
// Do Stuff
// Open/close Sockets etc
}
Election.setMessageFlag(false);
synchronized (Election.lock) {
Election.lock.notifyAll();
}
也就是说，多个线程可能会在设置标志之前(在下一行!)传递第一个 if 语句，以使后续线程等待。
但是，如果您使用 ReentrantLock，则应该使用如下代码:
lock.lock(); // block until condition holds
try {
// Do Stuff
// OPen CLose Sockets etc
} finally {
lock.unlock()
}
显然，您可能仍然需要设置一些标志来说明选举是否正在进行，但请确保使用真正的锁来保护正在运行的代码，例如ReentrantLock，或在适当的同步块(synchronized block)内。
希望这有帮助
关于java - Bully 算法的可疑代码输出，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/10210516/