The receive method is shown below:. The receive method handles all requests and responses. Upon reception of a message, we need to grab its byte[] buffer, unmarshal it into a Request and then handle the request. We use the JGroups helper method Util. On reception of RESULT sent by a slave , we set the result in the promise, releasing the blocked submitter of the task. The handleExecute method checks if a node should accept the task and, if yes, passes it to a thread pool to execute:.
First, we add the task to our tasks cache, keyed by the ClusterID5. This is the rank of the node which should execute the task. If it matches our own rank, we create a Handler and pass it to the thread pool to be executed on a separate thread, otherwise we return from handleExecute.
The Handler class is shown below:. It executes the task against the Slave interface handle method and stores the result. If there is an exception, then the exception which is serializable by default is stored as result instead.
Then a Response object is created from the result. Our code is now almost complete. The only thing missing is the handling of membership changes. Remember that we need to resubmit tasks from failed nodes, or from nodes who left gracefully, to other nodes. This is done in viewAccepted View :. First, we determine which members left between the new and previous views. This is done with Util.
Then we set the local address Channel. Next, the rank is computed by iterating through the new membership and comparing each element against the local address.
On a match, our rank is the iteration counter. Finally, we need to determine whether any nodes left since the previous view, and whether there are any tasks to take over from them. This is done by iterating through all left members if there are any and calling handleLeftMember , which is shown below:. This method iterates through all cache entries and compares the ID modulo cluster size to our own rank. If it matches, we execute the task unless the submitter itself left, in which case we drop the task.
Both assignments need to happen before handleLeftMember is called, as this method uses the 2 variables. If C crashes, D's and E's ranks change: D's rank is now 2. This means that D will process all of the tasks that C was processing and which hadn't completed by the time C crashed otherwise C would have removed them.
This means that B, C and D will now execute tasks which were already being worked on by other nodes. For example, C will re-execute D's tasks and B will re-execute C's tasks. This is not incorrect, as the submitter of a task will remove the task when completed. So, when receiving a result R from a slave for a task which was already completed and therefore removed, the submitter just drops R. This is not wrong, but leads to spurious and unneeded processing. A better way to define the rank would be to use consistent hashing [2] , which minimizes changes to the rank and therefore reexecution of tasks already being worked on by other nodes.
Our code is now complete. The last thing to do is to write the driver code, which we also add to Server:. The main method creates a Server and starts it. The loop method waits for a key press and then submits a short running on '1' or long running on '2' task. The task simply returns a new Date with the current time.
The long running task sleep for 15 seconds before returning the date. When 'q' is pressed, we stop the server gracefully and return. Let's see whether this thing works! The demo JAR can be downloaded here. Let's start some instances and submit some tasks.
To start an instance, we run:. Replace the IP address set with -Djgroups. We can see that we're the first node in the cluster, our local address is When we submit a task, we see that it is executed by our self, since we're the only node in the cluster:. We can see that the view now has 2 members: Note that for this demo, we start all instances as separate processes on the same host, but of course we would place those processes on different hosts in real life.
If we now go back to the first instance and submit 2 tasks, we can see that they are assigned to both instances:. Task 2 was executed by our self, but task 3 was executed by the second instance this can be verified by looking at the output of the second instance.
Let's now start a third instance:. We see that the cluster now has 3 nodes, and the rank of the freshly started instance is 2. Now we'll submit a long running task T and - before T completes - kill the node which is processing T.
Let's submit that task on the third instance:. Before the 15 seconds elapse, let's kill the second instance. After a few seconds, the output of the third instance shows the following:. This might be somewhat surprising, but correct. Let's see what's happening. First we get a view change, the new view is Therefore when task 1 is reassigned, it is the third node If we had more nodes in the cluster, the likelihood of a submitter processing its own task would decrease.
We implemented a simple, highly decentralized, clustered task distribution system in roughly lines of code and 5 classes. The system is failure resilient, because all nodes are peers and there's no central server. All peers are equal every peer can act as both master and slave and tasks are grabbed by a node based on an ID assigned by the submitter master.
Crashes or graceful termination of nodes doesn't lead to missing tasks, as the system re-balances itself and assigns orphaned tasks to another node in the cluster. The system is so small because it runs on top of JGroups. Had we written it without JGroups, we would have had to implement the following functionality ourselves:. The current task distribution is far from complete after all, this is just a demo of what can be done with JGroups!
The full demo code can be downloaded here. Thanks for visiting DZone today,. JGroups is a toolkit for reliable messaging. It can be used to create clusters whose nodes can send messages to each other. The main features include Cluster creation and deletion. The most powerful feature of JGroups is its flexible protocol stack, which allows developers to adapt it to exactly match their application requirements and network characteristics.
The instances are listed in order of joining the cluster, with the oldest instance as first element. Sending messages is now as simple as typing a message after the prompt and pressing return. The message will be sent to the cluster and therefore it will be received by both instances, including the sender. When "exit" or "quit" is entered, then the instance will leave the cluster.
This means, a new view will be installed immediately. To simulate a crash, simply kill an instance e. The other surviving instance will receive a new view, with only 1 instance itself and excluding the crashed instance.
One of the use cases of JGroups is to maintain state that is replicated across a cluster. For example, state could be all the HTTP sessions in a web server. Any update to a session is replicated across the cluster, e. This is needed so that all servers have the same state. However, what happens when a new server is started? That server has to somehow get the state e.
This is called state transfer. Note that, in order to be able to use state transfer in an application, the protocol stack has to have a state transfer protocol the default stack used by the demo app does. If the state cannot be transferred within this time, then an exception will be thrown.
ReceiverAdapter defines a callback getState which is called on an existing instance usually the coordinator to fetch the cluster state. In our demo application, we define the state to be the chat conversation. This is a simple list, to the tail of which we add every message we receive.
Note that this is probably not the best example for state, as this state always grows. As a workaround, we could have a bounded list, which is not done here though. The getState method is called in the state provider , ie. It is passed an output stream to which the state has to be written. Since access to state may be concurrent, we synchronize it.
Then we call Util. The setState method is called on the state requester , ie. Its task is to read the state from the input stream and set it accordingly:.
We again call a JGroups utility method Util. We also print the number of messages in the received chat history to stdout.
Note that this is not feasible with a large chat history, but - again - we could have a bounded chat history list. In this tutorial, we showed how to create a channel, join and leave a cluster, send and receive messages, get notified of view changes and implement state transfer. Building blocks are classes residing on top of a JChannel that provide a higher abstraction level, e. The protocol stack allows for complete customization of JGroups: protocols can be configured, removed, replaced, enhanced, or new protocols can be written and added to the stack.
Copyright Red Hat - Installation 1. Download JGroups can be downloaded here. JGroups core, demo and selected test classes Sample configuration files, e. JGroups 4. Final for example. Configuration Add jgroups Testing your Setup To see whether your system can find the JGroups classes, execute the following command:.
0コメント