passing data between threads

[abdullahansari]

Hello,
I would like to know what is the safest way of communicating between threads in java.
I have four threads that collect udp packets on different ports. I would like the data from all threads written to file through a single thread.
I used a simple communication mechanism using shared object and booleans to wait and exit the synchronized method but this does not seem to work all the time.

I would like to know if there is a safe way of passing arrays between threads and communicating.

[doWhile]

Define 'doesn't work all the time'...you should synchronize on the object/methods accessed between the threads (no example code, no way to tell what exactly is going on). If you wish to write out to a file as you describe, create a thread responsible for this task which contains a queue of the data to write - other threads add to the queue (in a synchronized manner). This writer thread can then wait (if queue is empty) and get notified when it has something to write, then proceed through each object in the queue, writing the contents to the file.

[abdullahansari]

Thank you for the reply.
Here is the code that i am working with.
I am very new to java.

----shared class for all the threads
public class Drop
{



private boolean emptyone = true;
private boolean emptytwo = true;
private boolean emptythree = true;
private boolean emptyfour = true;

private byte[] firstcallarray;
private byte[] secondcallarray;
private byte[] thirdcallarray;
private byte[] fourcallarray;

private int counterone = 1;
private int countertwo = 1;
private int counterthree = 1;
private int counterfour = 1;



private boolean doneone = true;
private boolean donetwo = true;
private boolean donethree = true;
private boolean donefour = true;


private byte[][] filearray = new byte[12][] ;
byte[] result;

private int SAfirst;
private int SAsecond;
private int SAthird;
private int SAfourth;
private int[] SAresult = new int[4];

private byte[] smone;
private byte[] smtwo;
private byte[] smthree;
private byte[] smfour;
private byte[][] counterarray = new byte[4][];

private byte[] sstone;
private byte[] ssttwo;
private byte[] sstthree;
private byte[] sstfour;
private byte[][] counterarraytwo = new byte[4][];


public synchronized byte[][] take()
{

while (emptyone || emptytwo || emptythree || emptyfour)
{
try
{
wait();
}
catch (InterruptedException e)
{

}
}
System.out.println("result");
filearray[0] = firstcallarray;
filearray[1] = secondcallarray;
filearray[2] = thirdcallarray;
filearray[3] = fourcallarray;


filearray[4] = smone;
filearray[5] = smtwo;
filearray[6] = smthree;
filearray[7] = smfour;

filearray[8] = sstone;
filearray[9] = ssttwo;
filearray[10] = sstthree;
filearray[11] = sstfour;



doneone = true;
donetwo = true;
donethree = true;
donefour = true;
emptyone = true;
emptytwo = true;
emptythree = true;
emptyfour = true;
notifyAll();

return filearray;
}










public synchronized void putone(byte[] x,int SA1,byte[] mone, byte[] stone)
{


while(doneone == false)
{
try
{
wait();
}
catch (InterruptedException e)
{

}

}

firstcallarray = x;
SAfirst = SA1;
smone = mone;
sstone = stone;
doneone = false;


emptyone = false;
notifyAll();

}


public synchronized void puttwo(byte[] y,int SA2,byte[] mtwo, byte[] sttwo)
{

while(donetwo == false)
{
try
{
wait();
}
catch (InterruptedException e)
{

}

}


secondcallarray = y;
SAsecond = SA2;
smtwo = mtwo;
ssttwo = sttwo;
donetwo = false;
emptytwo = false;
notifyAll();

}


public synchronized void putthree(byte[] z,int SA3,byte[] mthree, byte[] stthree)
{
while(donethree == false)
{

try
{
wait();
}
catch (InterruptedException e)
{

}

}


thirdcallarray = z;
SAthird = SA3;
smthree = mthree;
sstthree = stthree;
donethree = false;
emptythree = false;


notifyAll();

}


public synchronized void putfour(byte[] w,int SA4, byte[] mfour, byte[] stfour)
{


while(donefour == false)
{
try
{
wait();
}
catch (InterruptedException e)
{

}

}

fourcallarray = w;
SAfourth = SA4;
smfour = mfour;
sstfour = stfour;
donefour = false;
emptyfour = false;
notifyAll();

}
}







-----------first thread.

import java.lang.System;
import java.lang.String;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.io.*;
import java.net.*;


public class Producerone implements Runnable {

private Drop drop;
InetAddress address;
InetAddress destination;
String s = "192.168.0.1";
String t = "192.168.0.2";

DatagramSocket socket;
DatagramSocket socket1;
String str;
String str1;
int countfirst;
int j;
byte []z = new byte[2000000];
short array;
long newxfullstore ;
long oldxfullstore ;

long differenceone;
byte[]m = new byte[8000];
byte[]st = new byte[8000];
int k;
int no;
int q;
int kt;
byte[]y = new byte[1008];
byte[]newx = new byte[4];
byte[]oldx = {1,2,3,4};
byte[]newxfull = new byte[4];
byte[]oldxfull = {1,2,3,4};
byte[]yfull = new byte[1008];
byte []MA = new byte[8000000];
int SA = 0;
int QA = 0;
byte []RA;

int i;
boolean n;
boolean nfullfirst;
FileOutputStream first;
FileOutputStream second;
FileOutputStream third;
public Producerone(Drop drop) throws Exception
{
this.drop = drop;
address = InetAddress.getByName(s);
destination = InetAddress.getByName(t);

socket = new DatagramSocket(1026,address);
socket1 = new DatagramSocket(1042,address);

countfirst=0;
j = 1000;
newxfullstore = 0;
oldxfullstore = 0;
y[0] = 1;
y[1] = 2;
y[2] = 3;
y[3] = 4;
yfull[4]= 0;
yfull[5]= 0;
yfull[6]= 0;
yfull[7]= 0;
k = 0;
no = 0;
q = 0;
kt = 0;
nfullfirst = false;
differenceone = 0;

}


public void run()
{

while(true)
{


while(countfirst < 2000000)
{
DatagramPacket packet = new DatagramPacket(new byte[1008],1008);
try
{
socket.receive( packet );
socket.setSoTimeout(1000000);
}
catch(SocketTimeoutException timeExc)
{
System.exit(0);
}
catch (IOException e) {
System.out.println("socket cannot be opened");

}
countfirst = countfirst + 1000;
-----------i do bunch of stuff here which are not included to get
------------ MA,SA,m,st for 2000000 .
}

}

j=j+1;
countfirst = 0;

-----puts the array MA,m,st in to the shared class synchronized method...
---******** only access to shared class method*****-------
drop.putone(MA,SA,m,st);
SA = 0;
no = 0;
}
}
}


--- The other three threads are same except instead of drop.putone , i have
drop.puttwo, drop.putthree.










--- Here is the writer thread that writes to a file






import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;



public class Consumer implements Runnable
{
private Drop drop;
FileOutputStream first;
FileOutputStream second;
FileOutputStream third;
FileOutputStream fourth;
FileOutputStream firstcount;
FileOutputStream secondcount;
FileOutputStream thirdcount;
FileOutputStream fourthcount;
FileOutputStream firstScount;
FileOutputStream secondScount;
FileOutputStream thirdScount;
FileOutputStream fourthScount;
int j = 1000;
//bis byte [][] res = new byte[4][];
byte [][]res = new byte[12][];
int []resultlength = new int[4];
public Consumer(Drop drop)
{
this.drop = drop;
}

public void run()
{
while(true)
{
try
{
first = new FileOutputStream(new File("C://test1//"+ Integer.toString(j)+".txt"),true);
second = new FileOutputStream(new File("C://test2//"+ Integer.toString(j)+".txt"),true);
third = new FileOutputStream(new File("C://test3//"+ Integer.toString(j)+".txt"),true);
fourth = new FileOutputStream(new File("C://test4//"+ Integer.toString(j)+".txt"),true);
//bis java counters
firstcount = new FileOutputStream(new File("C://test5//"+ Integer.toString(j)+".txt"),true);
secondcount = new FileOutputStream(new File("C://test6//"+ Integer.toString(j)+".txt"),true);
thirdcount = new FileOutputStream(new File("C://test7//"+ Integer.toString(j)+".txt"),true);
fourthcount = new FileOutputStream(new File("C://test8//"+ Integer.toString(j)+".txt"),true);
//bis memory counters.
firstScount = new FileOutputStream(new File("C://test9//"+ Integer.toString(j)+".txt"),true);
secondScount = new FileOutputStream(new File("C://test10//"+ Integer.toString(j)+".txt"),true);
thirdScount = new FileOutputStream(new File("C://test11//"+ Integer.toString(j)+".txt"),true);
fourthScount = new FileOutputStream(new File("C://test12//"+ Integer.toString(j)+".txt"),true);
}
catch (FileNotFoundException e)
{
System.out.println("Test folder not found");
}



---- ********only access to shared class method *********.
res = drop.take();


try
{
first.write(res[0],0,resultlength[0]);
second.write(res[1],0,resultlength[1]);
third.write(res[2],0,resultlength[2]);
fourth.write(res[3],0,resultlength[3]);

firstcount.write(res[4]);
secondcount.write(res[5]);
thirdcount.write(res[6]);
fourthcount.write(res[7]);

firstScount.write(res[8]);
secondScount.write(res[9]);
thirdScount.write(res[10]);
fourthScount.write(res[11]);


} catch (IOException e)
{
System.out.println("cannot write file");
}

j = j + 1;


try
{
first.close();
second.close();
third.close();
fourth.close();

firstcount.close();
secondcount.close();
thirdcount.close();
fourthcount.close();

firstScount.close();
secondScount.close();
thirdScount.close();
fourthScount.close();
} catch (IOException e)
{

System.out.println("cannot close the file");
}

}

}
}


"Does not work all the time" ...i am trying to write same data in all the files but when i try to read out the data from the file through separate program some of the data that is read out is not the same in all the files even though most of the times it is the same.

I would really appreciate if you can help me out .
I dont know about blocking queue...if you can suggest any tutorial i would read it.