Stanley Walker

Stephan van Hulst wrote:The fact that it works at all is more luck than anything else.

A big part of the problem is that you're passing the same list instance to every asynchronous task handler and clearing it instead of creating a new one. So not only does every handler work with the same list of lines, the lines are cleared at the same time that the processing begins.

thank you stephan for your response.  i actually corrected that piece using

processChunk(new ArrayList<String>(chunk));

I appreciate your code sample, pretty impressive.
however my main question is regarding join. will completablefutures not run unless join is called?

I wanted to implement a functionality where a big file gets broken down into chunks and processing can happen in parallel.

I used CompletableFuture to run tasks in parallel. unfortunately , it doesnt work unless i use join. Im surprised that this is happening, since according to docs, get is also a blocking methd in the class. can someone please help me in figuring out what  i am doing wrong


package com.org.java_trial.thread.executors; import java.io.BufferedReader; import java.io.FileReader; import java.io.IOException; import java.nio.file.Files; import java.nio.file.Paths; import java.util.ArrayList; import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import com.org.java_trial.thread.bigfile.ReadAndProcessBigFile; public class ProcessReallyBigFile { private static final ExecutorService ex = Executors.newFixedThreadPool(5); private static CompletableFuture<String> processChunk(List<String> lines) { CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> { //just for purposes of testing, will be replaced with major function later lines.stream().forEach(System.out::println); return "done"; }, ex); //cf.join(); if i uncommnet this everything works return cf; } private static void readInChunks(String filepath, Integer chunksize) { List<CompletableFuture<String>> completable = new ArrayList<>(); try (BufferedReader reader = Files.newBufferedReader(Paths.get(filepath))) { String line = null; List<String> collection = new ArrayList<String>(); int count = 0; while ((line = reader.readLine()) != null) { if (count % chunksize == chunksize - 1) { collection.add(line); completable.add(processChunk(collection)); collection.clear(); } else { collection.add(line); } count++; } // any leftovers if (collection.size() > 0) completable.add(processChunk(collection)); } catch (IOException e) { e.printStackTrace(); } for (CompletableFuture c : completable) { c.join(); if (c.isDone() || c.isCompletedExceptionally()) { try { System.out.println("i cmpleted" + c.get()); } catch (InterruptedException | ExecutionException e) { // TODO Auto-generated catch block e.printStackTrace(); } } } ex.shutdown(); } public static void main(String[] args) { String filepath = "C:\\somak\\eclipse-workspace\\java_thingies\\java_trial\\account_1.csv"; readInChunks(filepath, 3); } }


my file contents is pretty small:

1212451,London,25000,Blocked
1212452,London,215000,Open
1212453,London,125000,CreditBlocked
1212454,London,251000,DebitBlocked
1212455,London,2500,Open
1212456,London,4000,Closed
1212457,London,25100,Dormant
1212458,London,25010,Open
1212459,London,27000,Open
12124510,London,225000,Open
12124511,London,325000,Open
12124512,London,425000,Open
12124513,London,265000,Open
12124514,London,2577000,Open
12124515,London,2504400,Open

//cf.join(); if i uncommnet this everything works

in case i uncomment the above line in the method processChunk, everything works fine. my values are printed and everything. however if i remove it, nothing happens. all i get are notifications that futures have copleted but the contents are not printed


This is my output
i cmpleteddone i cmpleteddone i cmpleteddone i cmpleteddone i cmpleteddone

Hi,

other than implementing observer pattern, is there any other way to pass a message between a contained object to a container object?


Essentially, lets say i have a Computer class which contains a single SoundCard class. Essentially Computer has-a Soundcard.

now i Computer object can call methods on SoundCard objects, by passing a parameter -- essentially pass a message to the soundcard object.

but how does the other way happen? I dont want Soundcard to hold a reference to the computer object.

so how can soundcard pass a message to the Computer object.

one way i can think of is observer pattern. is there any other way?

sample output of the above code is

i am in doing smthing dont distubr 2018-06-25T21:51:49.623
i am in doing smthing dont distubr 2018-06-25T21:51:54.626
i am in doing smthing dont distubr 2018-06-25T21:51:59.628
i am in doing smthing dont distubr 2018-06-25T21:52:04.630
----clear waiting time...i am able to pause and resume-------
i am in doing smthing dont distubr 2018-06-25T21:52:29.558
i am in doing smthing dont distubr 2018-06-25T21:52:34.558
i am in doing smthing dont distubr 2018-06-25T21:52:39.559

Hi,

I want to implement a service, where some work happens periodically in its own thread.
However outside client should be able to pause the thread for a while, and then resume it

I have been able to implement the same using ScheduledExcutorService and semaphore. however i am not satisfied with the implementation. As you can see in the code below, it has become necessar for the client to pass a semaphore instance to the running task.

is there a better way to implement the same

package com.org.pausable; import java.time.LocalDateTime; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.stream.IntStream; class MyOwnPausable { private Semaphore semaphore; public MyOwnPausable(Semaphore semaphore) { this.semaphore = semaphore; } public void compute() throws InterruptedException { semaphore.acquire(); System.out.println("i am in doing smthing dont distubr " + LocalDateTime.now()); semaphore.release(); } } public class SomakPausableThread { public static void main(String[] args) throws InterruptedException { Semaphore semaphore = new Semaphore(1); ScheduledExecutorService service = Executors.newScheduledThreadPool(2); service.scheduleWithFixedDelay(() -> { try { new MyOwnPausable(semaphore).compute(); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } }, 0, 5, TimeUnit.SECONDS); Thread.sleep(20000); semaphore.acquire(); Thread.sleep(20000); semaphore.release(); Thread.sleep(100000); service.shutdown(); } }

hi Junilu

thank you so much for your responses.
this is realy helpful

Piet Souris wrote:hi Stanley,

as said: it is not a question of whether a Function works and a Consumer doesn't, but the question what is a legal Function and what is a legal Consumer.

I must say: your code, although nothing wrong with it, is not very transparant. To make the error a little easier to spot, let's make a separate Function and a separate Consumer. If you get an error in one of these, then you know that the problem is not in the 's.changeState(2)' part of the code,

So here goes:
Function<StateCount, Integer> sToi = c -> c.getCount() + 1;
So, when we apply this function to some StateCount c, then the return value is c.count + 1.
On its own, c.count + 1 means nothing, you get a compiler error here, stating that this value should be assigned to a variable. Now have a look at the method 'stateChanged':
public void changeState(Function<StateCount, Integer> consumer) {    this.count = consumer.apply(this);    System.out.println(this.count); }
You invoke it with:
StateCount s = new StateCount(0); s.changeState((c) -> c.getCount() + 10);
So, in effect, what you do is saying:
this.count = this.getCount() + 1; System.out.println(this.count);
and as you can see, this is perfectly legal code.

Now for the Consumer part. Again, let's check if the Consumer as you defined it is correct:
Consumer<StateCount> consumer = s -> s.getCount() + 1;
As you will notice, this gives a compiler error, stating that an int (the result of the right part) cannot be converted to void, as a Consumer must be.
So, let's try a variant:
Consumer<StateCount> consumer = s -> {s.getCount() + 1;};
Now we return nothing, but as before, we get the error that the right hand part is not a statement and should be assigned to a variable.
Well, let's do that, but the way I do it makes it necessary to make the variable 'count' available to other classes as well: make it a public variable (that is bad practice, never do that in your code, but it is just for demonstrations sake)
Consumer<StateCount> consumer = s -> s.count = s.getCount() + 1;
And that is perfectly legal as well.

So, with the Function sToi and Consumer consumer defined (and legal), we can invoke them, either as you invoke them,but also as follows:
StateCount s = new StateCount(0); s.changeState(sToi); s.changeState2(consumer);
As said, having the separate sToi and consumer makes it just a tad easier to see what is going on.

Looking back, I must admit that this is actually not an easy subject at all! It is a subject that takes a lot of practising, and a lot of trying until you get it right! But the reward of putting all that effort is there: it is big fun working with all of these Consumers, Functions, Predicates and similar. Of course, if you have additional questions, just come up with them.

THank you Piet, i totally appreciate the pain you took to explain the problem to me. And i must say, i do understand now why one works and the other doesnt.

Appreciate your help

Junilu Lacar wrote:Additionally, your usage of the Function and Consumer interfaces are smelly. I can't quite put my finger on it but it just seems like there's way too much unnecessary intimacy between the StateCount class and the lambdas. The lambdas having a reference to StateCount does not look right to me. And Consumer seems like the opposite of what you should use. That is, it seems to me a Producer Supplier is more appropriate for what you want to do. I'll try to come up with an example to show something I think is more sensible.

Hi Junilu,

I promise you, i was just trying out different ways to fail. to be honest, thats the only way i have learnt lambdas as much as  i have(which is not too much, i must confess)
i keep trying out weird stuff with functions api and this particular piece confused me, since one works and the other doesnt. and compiler errors are not that descriptive either

Can i ask you a question - why do you say  

Junilu Lacar wrote:between the StateCount class and the lambdas

Is it because the lambdas passed in is changing the state of the class??

Piet Souris wrote:hi Stanley,

it is the expresson 'c -> s.getCount() += 1' that is invalid. In fact, it boils down to: s.getCount() = s.getCount() + 1, which is invalid.

If you add the method
public void increase() {    count = getCount() + 1;   // normally you would simply do: count += 1 }
and make the Consumer:
s.changeState2(c -> c.increase());
then all is well.

Hi Piet,

Thank you for the response.

I understand  s.getCount() = s.getCount() + 1 is an invalid statement

But why does it work when i use a Function

check here
public void changeState(Function<StateCount, Integer> consumer) { this.count = consumer.apply(this); System.out.println(this.count); } StateCount s = new StateCount(0); s.changeState((c) -> c.getCount() + 10);

the above code works. How come consumer it doesnt work

like below

public void changeState2(Consumer<StateCount> consumer) { consumer.accept(this); System.out.println(this.count); } StateCount s = new StateCount(0); s.changeState2((StateCount c) -> c.getCount() + 1);

essentially, i am not able to figure out why FUnction works and consumer does not

HI,

I want to create a simple class which has a single instance variable int count.
I want to be able to add/decrease the value of count. But i dont want to have methods increment/decrement. i want a single method which does the same by taking a lambda


if you see the below code, when i use a FUnction it works.
however if i use a consumer , i am not able to do it using a consumer.

i get this error. The left-hand side of an assignment must be a variable and The method changeState1(Function<StateCount,Integer>) is ambiguous for the type StateCount

import java.util.function.Consumer; import java.util.function.Function; class StateCount { private int count; public StateCount(int count) { this.count = count; } public void changeState(Function<StateCount, Integer> consumer) { this.count = consumer.apply(this); System.out.println(this.count); } public void changeState2(Consumer<StateCount> consumer) { consumer.accept(this); System.out.println(this.count); } public int getCount() { return this.count; } } public class IncrementOrDecrementStateCount { public static void main(String[] args) { StateCount s = new StateCount(0); s.changeState((c) -> c.getCount() + 10); s.changeState((c) -> c.getCount() - 1); s.changeState((c) -> c.getCount() + 1); s.changeState((c) -> c.getCount() + 1); s.changeState((c) -> c.getCount() + 1); s.changeState2((c) -> c.getCount() += 1); } }

java.util.function.Predicate has a method isEquals with the following signature:

static <T> Predicate<T> isEqual(Object targetRef)

i do not understand the signature though. THe method returns Predicate<T> . i get that , however what is <T> Predicate<T> . why the additional <T> in the front?

Really looking forward to any help with the above question

I have been playing around FutureTask recently. I wanted to fetch data from two tables in a parallel execution. Currently my solution is mocked, where it just fetches some hardcoded data.

1. I created futuretask1 which fetches data from 1st table
2. I created futuretask2 which fetches data from 2nd table
3. I submitted the task to executor services for execution. Things are working fine.
4. However my problem is if I submit any one of the future task twice, i do not see execution happening twice. Clearly java is not executing a task which it has already executed. However I could not find anything alluding to it in the java docs for ExecutorService or Executor. I had also opened up the java api, and I could not figure out how it detects uniqueness.

Clearly I am missing out something here, any help would be appreciated.
My code snippets are given below:

package com.org.somak.thread.future; import java.math.BigDecimal; import java.util.HashSet; import java.util.Set; import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.FutureTask; public class MockedFetchFromTable { public static void main(String[] args) { parallelFetchFromTwoTable(); } private static void parallelFetchFromTwoTable() { FutureTask<DataFromFirstTable> futureTask = new FutureTask<DataFromFirstTable>( new Callable<DataFromFirstTable>() { @Override public DataFromFirstTable call() throws Exception { return loadDataFromFirstTable(); } }); FutureTask<DataFromSecondTable> futureTask2 = new FutureTask<DataFromSecondTable>( new Callable<DataFromSecondTable>() { @Override public DataFromSecondTable call() throws Exception { return loadDataFromSecondTable(); } }); ExecutorService exe = Executors.newCachedThreadPool(); //ExecutorService exe = Executors.newFixedThreadPool(1); exe.submit(futureTask); exe.submit(futureTask2);//submitting task three times in total in next lines of code exe.submit(futureTask2); exe.execute(futureTask2); exe.shutdown(); try { DataFromFirstTable d = futureTask.get(); DataFromSecondTable d1 = futureTask2.get(); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } } private static DataFromFirstTable loadDataFromFirstTable() { System.out.println("MockedFetchFromTable.loadDataFromFirstTable()"); DataFromFirstTable d = new DataFromFirstTable(); d.setAmount(BigDecimal.TEN); d.setCurrency("AUD"); System.out .println("MockedFetchFromTable.parallelFetchFromTwoTable() result----->" + d); return d; } private static DataFromSecondTable loadDataFromSecondTable() { System.out.println("MockedFetchFromTable.loadDataFromSecondTable()"); DataFromSecondTable d1 = new DataFromSecondTable(); d1.setFirstName("first name"); d1.setNameId("nameId"); d1.setSecondName("secondName"); System.out .println("MockedFetchFromTable.parallelFetchFromTwoTable() second result----->" + d1); return d1; } }

The classes DataFromFirstTable and DataFromSecondTable are simple POJO's

package com.org.somak.thread.future; import java.math.BigDecimal; public class DataFromFirstTable { private BigDecimal amount; private String currency; public BigDecimal getAmount() { return amount; } public void setAmount(BigDecimal amount) { this.amount = amount; } public String getCurrency() { return currency; } public void setCurrency(String currency) { this.currency = currency; } @Override public String toString() { return this.getCurrency() + "#" + this.getAmount(); } }


package com.org.somak.thread.future; public class DataFromSecondTable { private String nameId; private String firstName; private String secondName; public String getNameId() { return nameId; } public void setNameId(String nameId) { this.nameId = nameId; } public String getFirstName() { return firstName; } public void setFirstName(String firstName) { this.firstName = firstName; } public String getSecondName() { return secondName; } public void setSecondName(String secondName) { this.secondName = secondName; } @Override public String toString() { return this.getNameId() + "#" + this.getFirstName() + "#" + this.getSecondName(); } }


The output that i get on running the above code is:

MockedFetchFromTable.loadDataFromFirstTable()
MockedFetchFromTable.loadDataFromSecondTable()
MockedFetchFromTable.parallelFetchFromTwoTable() second result----->nameId#first name#secondName
MockedFetchFromTable.parallelFetchFromTwoTable() result----->AUD#10

The output tells me that even though I have submitted my futuretask2 twice, the line MockedFetchFromTable.loadDataFromSecondTable() is only getting printed once.

Please help.

Also one more question, if i do not shutdown my executorservice manually , i can see that my program waits for close to a minute even after all execution is complete. Is this normal behavior? or have i done something wrong?

HI,
This is a question one of my juniors asked me and I was completely dumbstruck.

according to java docs for Hibernate :

for Session

It is not intended that implementors be threadsafe. Instead each thread/transaction should obtain its own instance from a SessionFactory.

for SEssionFactory.getCurrentSEssion()

Obtains the current session. The definition of what exactly "current" means controlled by the CurrentSessionContext impl configured for use.

So if session object is not threadsafe, what happens when two threads call sessionFactory.getCurrentSession() .... if they get the same session object will there not be chances of all the dangers that threads cause when they get the same instance?

java api of CurrentSessionContext did not help and when i browsed through the code for implementors of CurrentSessionContext, it seemed to me that it is returning the same session object.

Please help, this has been bugging me for some time now.