https://medium.com/building-the-open-data-stack/observer-versus-pub-sub-design-patterns-48b1dbc83916
Author Archives: admin
Queries
How to Delete duplicate Rows from Table
In the below table studId 1,6 and 9 is repeated which should be deleted.
studId | studentName | age |
---|---|---|
1 | Mugil | 35 |
2 | Vinu | 36 |
3 | Viju | 42 |
4 | Mani | 35 |
5 | Madhu | 36 |
6 | Mugil | 35 |
7 | Venu | 37 |
8 | Banu | 34 |
9 | Mugil | 35 |
Below query wont work on MySQL but the format doesn’t change. When you take Max only last occurrence of row would be taken and others would be excluded.
DELETE FROM tblStudents TS WHERE TS.studId NOT IN (SELECT MAX(TSS.studId) FROM tblStudents TSS GROUP BY TSS.studentName, TSS.age)c
The same could be done using MIN function.
SELECT TS.* FROM tblStudents TS WHERE TS.studId NOT IN (SELECT MIN(TSS.studId) FROM tblStudents TSS GROUP BY TSS.studentName, TSS.age)
Output
studId | studentName | age |
---|---|---|
6 | Mugil | 35 |
9 | Mugil | 35 |
Threads Programs – Interview Questions
Simple Program using Runnable and Callable
HelloThread1.java
public class HelloThread1 implements Runnable{ @Override public void run() { System.out.println("Hello World from Thread Name (" + Thread.currentThread().getName() +") using Runnable "); } }
HelloThread2.java
public class HelloThread2 implements Callable { @Override public Object call() throws Exception { return "Hello World from Thread Name (" + Thread.currentThread().getName() +") using Callable"; } }
Main.java
public class Main { public static void main(String[] args) throws Exception { HelloThread1 objThread1 = new HelloThread1(); //Instance of Runnable HelloThread2 objThread2 = new HelloThread2(); //Instance of Callable objThread1.run(); System.out.println(objThread2.call()); } }
Output
Hello World from Thread Name (main) using Runnable Hello World from Thread Name (main) using Callable
Simple Program to print numbers using threads
NumberPrinter.java
public class NumberPrinter implements Runnable{ int number; public NumberPrinter(int number){ this.number = number; } public void run(){ System.out.println("Printing Number from Thread "+ this.number); } }
Main.java
public class Main { public static void main(String[] args) { for (int idx=1;idx<=5;idx++){ Thread objthread = new Thread(new NumberPrinter(idx)); objthread.start(); } } }
Output
Printing Number from Thread 5 Printing Number from Thread 1 Printing Number from Thread 4 Printing Number from Thread 3 Printing Number from Thread 2
Simple Program using Executor Service taking Runnable as Argument
ExecutorService is a framework which allows to create thread. Threads can be created from FixedThreadPool, CachedThreadPool and ScheduledThreadPool. submit() method takes runnable or callable object (Functional Interface Type) as argument. The Same code above can be rewritten as below
Main.java
public class Main { public static void main(String[] args) throws Exception { ExecutorService objExecService = Executors.newFixedThreadPool(2); //Lambda Expresssion passed as Argument as Runnable is FI objExecService.submit(() -> { System.out.println(Thread.currentThread().getName()); }); objExecService.shutdown(); } }
Output
pool-1-thread-1
Same code with Runnable instance passed as argument to submit
. . //Instance of Runnable passed as argument HelloThread1 objHT1 = new HelloThread1(); objExecService.submit(objHT1); . .
Output
Hello World from Thread Name (pool-1-thread-1) using Runnable
Same code with Runnable as Anonymous Class passed as argument
ExecutorService exec = Executors.newFixedThreadPool(2); //Instance of Runnable passed as Anonymous class exec.execute(new Runnable() { public void run() { System.out.println("Hello world"); } }); exec.shutdown();
Simple Program using Executor Service taking Callable as Argument
public class Main { public static void main(String[] args) throws Exception { ExecutorService objExecService = Executors.newFixedThreadPool(2); Future<String> objFuture = objExecService.submit(new HelloThread2()); System.out.println(objFuture.get()); objExecService.shutdown(); } }
Output
Hello World from Thread Name (pool-1-thread-1) using Callable
Using Lambda Expression as Submi
. . ExecutorService objExecService = Executors.newFixedThreadPool(2); Future<String> objFuture = objExecService.submit(() -> { Thread.sleep(3000); return Thread.currentThread().getName(); }); System.out.println(objFuture.get()); . .
The above could be rewritten in anonymous class as below
ExecutorService objExecService = Executors.newFixedThreadPool(2); Future<String> objFuture = objExecService.submit(new Callable<String>() { @Override public String call() throws Exception { Thread.sleep(3000); return Thread.currentThread().getName(); } }); System.out.println(objFuture.get()); objExecService.shutdown();
Program for Creating Thread Pool and executing Task
ThreadPoolExample.java
public class ThreadPoolExample { public static void main(String args[]) { ExecutorService service = Executors.newFixedThreadPool(10); //create 10 worker threads in Thread Pool for (int i =0; i<100; i++){ service.submit(new Task(i)); //submit that to be done } service.shutdown(); } }
Task.java
final class Task implements Runnable { private int taskId; public Task(int id){ this.taskId = id; } @Override public void run() { System.out.println("Task ID : " + this.taskId +" performed by " + Thread.currentThread().getName()); } }
Task ID : 0 performed by pool-1-thread-1 Task ID : 3 performed by pool-1-thread-4 Task ID : 2 performed by pool-1-thread-3 Task ID : 1 performed by pool-1-thread-2 Task ID : 5 performed by pool-1-thread-6 Task ID : 4 performed by pool-1-thread-5
Threads and States
- NEW – a newly created thread that has not yet started the execution
- RUNNABLE – either running or ready for execution but it’s waiting for resource allocation
- BLOCKED – waiting to acquire a monitor lock to enter or re-enter a synchronized block/method
- WAITING – waiting for some other thread to perform a particular action without any time limit
- TIMED_WAITING – waiting for some other thread to perform a specific action for a specified period
- TERMINATED – has completed its execution
NEW Thread (or a Born Thread) is a thread that’s been created but not yet started.
It remains in this state until we start it using the start() method
NewState.java
public class NewState implements Runnable{ public void run(){ System.out.println("I am in new State"); } }
Main.java
public class Main { public static void main(String[] args) throws InterruptedException { Thread objThread = new Thread(new NewState()); System.out.println(objThread.getState()); } }
Output
NEW
Runnable When we’ve created a new thread and called the start() method on that, it’s moved from NEW to RUNNABLE state. Threads in this state are either running or ready to run, but
they’re waiting for resource allocation from the system. In a multi-threaded environment, the Thread-Scheduler (which is part of JVM) allocates a fixed amount of time to each thread. So it runs for a particular amount of time, then leaves the control to other RUNNABLE threads.
RunnableState .java
public class RunnableState implements Runnable{ public void run(){ System.out.println("I would be in Runnable State"); } }
Main.java
public class Main { public static void main(String[] args) throws InterruptedException { Thread objRThread = new Thread(new RunnableState()); objRThread.start(); System.out.println(objRThread.getState()); } }
Output
RUNNABLE I would be in Runnable State
This is the state of a dead thread. It’s in the TERMINATED state when it has either finished execution or was terminated abnormally.
TerminatedState.java
public class TerminatedState implements Runnable{ public void run(){ Thread objNewState = new Thread(new NewState()); objNewState.start(); } }
Main.java
public class Main { public static void main(String[] args) throws InterruptedException { Thread objTState = new Thread(new TerminatedState()); objTState.start(); objTState.sleep(1000); System.out.println("T1 : "+ objTState.getState()); } }
Output
I am in new State T1 : TERMINATED
A thread is in the BLOCKED state when it’s currently not eligible to run. It enters this state when it is waiting for a monitor lock and is trying to access a section of code that is locked by some other thread.
BlockedState.java
public class BlockedState implements Runnable{ public void run(){ blockedResource(); } public static synchronized void blockedResource(){ while(true){ //Do Nothing } } }
Main.java
public class Main { public static void main(String[] args) throws InterruptedException { Thread objB1Thread = new Thread(new BlockedState()); Thread objB2Thread = new Thread(new BlockedState()); objB1Thread.start(); objB2Thread.start(); Thread.sleep(1000); System.out.println(objB1Thread.getState()); System.out.println(objB2Thread.getState()); System.exit(0); } }
Output
RUNNABLE BLOCKED
A thread is in WAITING state when it’s waiting for some other thread to perform a particular action. According to JavaDocs, any thread can enter this state by calling any one of the following
object.wait() (or) thread.join() (or) LockSupport.park()
WaitingState.java
public class WaitingState implements Runnable{ public void run(){ Thread objWaitState = new Thread(new SleepState()); objWaitState.start(); try { objWaitState.join(); } catch (InterruptedException e) { throw new RuntimeException(e); } } }
SleepState.java
public class SleepState implements Runnable{ @Override public void run() { try { Thread.sleep(5000); } catch (InterruptedException e) { throw new RuntimeException(e); } } }
Main.java
public class Main { public static void main(String[] args) throws InterruptedException { Thread objWaitingThread = new Thread(new WaitingState()); objWaitingThread.start(); objWaitingThread.sleep(1000); System.out.println("T1 : "+ objWaitingThread.getState()); System.out.println("Main : "+Thread.currentThread().getState()); } }
Output
T1 : WAITING Main : RUNNABLE
A thread is in TIMED_WAITING state when it’s waiting for another thread to perform a particular action within a stipulated amount of time. According to JavaDocs, there are five ways to put a thread on TIMED_WAITING state:
thread.sleep(long millis) (or) wait(int timeout) (or) wait(int timeout, int nanos) thread.join(long millis) (or) LockSupport.parkNanos (or) LockSupport.parkUntil
TimedWaitState.java
public class TimedWaitState implements Runnable{ @Override public void run() { try { Thread.sleep(5000); } catch (InterruptedException e) { throw new RuntimeException(e); } } }
Main.java
public class Main { public static void main(String[] args) throws InterruptedException { Thread objTWState = new Thread(new TimedWaitState()); objTWState.start(); Thread.sleep(2000); System.out.println("T1 : "+ objTWState.getState()); } }
Output
T1 : TIMED_WAITING
Simple Ticket Reservation System
- We use ReentrantLock for locking the Resource(totalSeats)
- Incase anything goes wrong (Exception being thrown etc.) you want to make sure the lock is released no matter what.
- Calling the reserveSeats method should be done inside separate threads
ReservationSystem.java
import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; public class ReservationSystem { private Integer totalSeats; private final Lock lock = new ReentrantLock(); public ReservationSystem(Integer totalSeats){ this.totalSeats = totalSeats; } public Integer getTotalSeats(){ return totalSeats; } public void reserveSeats(String userName, int numOfSeats){ lock.lock(); try{ if(numOfSeats >0 && totalSeats>numOfSeats){ totalSeats -= numOfSeats; System.out.println(userName + " has reserved "+ numOfSeats + " with " + totalSeats + " still available"); }else{ System.out.println("Seats not Available"); } }finally { lock.unlock(); } } }
BookSeat.java
public class BookSeat { public static void main(String[] args) { ReservationSystem objResSys = new ReservationSystem(100); System.out.println("Total available Seats "+ objResSys.getTotalSeats()); Thread objThread1 = new Thread(() -> {objResSys.reserveSeats("User1", 10);}); Thread objThread2 = new Thread(() -> {objResSys.reserveSeats("User2", 20);}); Thread objThread3 = new Thread(() -> {objResSys.reserveSeats("User3", 5);}); objThread1.start(); objThread2.start(); objThread3.start(); try { objThread1.join(); objThread2.join(); objThread3.join(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } System.out.println("Remaining available Seats "+ objResSys.getTotalSeats()); } }
Total available Seats 100 User2 has reserved 20 with 80 still available User1 has reserved 10 with 70 still available User3 has reserved 5 with 65 still available Remaining available Seats 65
Simple Banking System handling Transactions with Threads
Banking System
- We have Bank Account with 2 Fields – balance and Account Number
- We have Transaction class implementing Runnable
- We create object for account with some initial balance and try to pass as parameter to runnable Transaction Object
BankAccount.java
import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; public class BankAccount { private Integer balance; private Integer accountNumber; private final Lock reLock = new ReentrantLock(); public BankAccount(Integer balance, Integer accountNumber){ this.balance = balance; this.accountNumber = accountNumber; } public void debitAmount(Integer amount){ reLock.lock(); try{ balance -= amount; }finally { reLock.unlock(); } } public void creditAmount(Integer amount){ reLock.lock(); try{ balance += amount; }finally { reLock.unlock(); } } public Integer getAccountNumber(){ return this.accountNumber; } public Integer getBalance(){ return this.balance; } }
BankTransaction.java
public class BankTransaction implements Runnable{ public Integer transAmount; public BankAccount bankAccount; public BankTransaction(Integer transAmount, BankAccount bankAccount){ this.transAmount = transAmount; this.bankAccount = bankAccount; } @Override public void run() { if(transAmount >= 0){ bankAccount.creditAmount(transAmount); }else{ bankAccount.debitAmount(Math.abs(transAmount)); } } }
BankSystem.java
public class BankSystem { public static void main(String[] args) { BankAccount objAcc1 = new BankAccount(1000, 101); BankAccount objAcc2 = new BankAccount(2000, 102); Thread objThread1 = new Thread(new BankTransaction(50, objAcc1)); Thread objThread2 = new Thread(new BankTransaction(-150, objAcc2)); Thread objThread3 = new Thread(new BankTransaction(250, objAcc2)); Thread objThread4 = new Thread(new BankTransaction(250, objAcc1)); objThread1.start(); objThread2.start(); objThread3.start(); objThread4.start(); try{ objThread1.join(); objThread2.join(); objThread3.join(); objThread4.join(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } System.out.println("Final Balance in Account " + objAcc1.getAccountNumber() + " with balance " + objAcc1.getBalance()); System.out.println("Final Balance in Account " + objAcc2.getAccountNumber() + " with balance " + objAcc2.getBalance()); } }
Output
Final Balance in Account 101 with balance 1300 Final Balance in Account 102 with balance 2100
Kafka Basics
Why Kafka?
One of the activity in application is to transfer data from Source System to Target System. Over period of time this communication becomes complex and messy. Kafka provides simplicity to build real-time streaming data pipelines and real-time streaming applications.
Factory Method Pattern
Define an interface for creating an object, but let subclasses decide which class to instantiate. Factory Method lets a class defer instantiation to subclasses.
- In Factory Pattern we have Product(Abstract), ConcreteProduct and Creator(Abstract), ConcreteCreator
- ConcreteCreator would create ConcreteProduct by implementing abstract factory method of Creator which has Product return type
- Incase if there is any new Product to be added it fully supports Open Closed Principle(Open For Extension, Closed for Changes).
- Open for Extension – Adding new ConcreteProduct and ConcreateCreator class, Closed for Changes – No changes in anyother code unlike Simple factory or static factory method which requires change in Switchcase, enum (or) if case
- Closed for Changes – No changes in anyother code unlike Simple factory or static factory method which requires change in Switchcase, enum (or) if case
JobProfile.java
abstract class JobProfile { public abstract String mandatorySkills(); public Integer defaultWorkHrs(){ return 8; } }
JavaProfile.java
public class JavaProfile extends JobProfile{ @Override public String mandatorySkills() { return "Java, Springboot, Microservices"; } }
SQLProfile.java
public class SQLProfile extends JobProfile{ @Override public String mandatorySkills() { return "Cosmos, MySQL, MSSQL"; } }
JobProfileCreator.java
abstract class JobProfileCreator { public JobProfile getJobProfile(){ JobProfile objJobProfile = createJobProfileFactory(); return objJobProfile; } public abstract JobProfile createJobProfileFactory(); }
JavaProfileCreator.java
public class JavaProfileCreator extends JobProfileCreator { @Override public JobProfile createJobProfileFactory() { return new JavaProfile(); } }
SQLProfileCreator.java
public class SQLProfileCreator extends JobProfileCreator { @Override public JobProfile createJobProfileFactory() { return new SQLProfile(); } }
Consultancy.java
public class Consultancy { public static void main(String[] args) { getProfileDetails(new JavaProfileCreator()); } public static void getProfileDetails(JobProfileCreator jobProfileCreator){ JobProfile objJobProfile = jobProfileCreator.getJobProfile(); System.out.println(objJobProfile.mandatorySkills() + " with "+ objJobProfile.defaultWorkHrs() + "hrs of Work"); } }
Output
Java, Springboot, Microservices with 8hrs of Work
Simple Factory
- In Simple Factory we have a Factory Class(LoggerFactory.java) and We call the createLogger method which returns different implementation of logger
- Logger is a abstract class which has different implementations
Logger.java
public abstract class Logger { abstract void log(String logstring); }
ConsoleLogger.java
public class ConsoleLogger extends Logger{ @Override void log(String logstring) { System.out.println("Logging to Console - "+ logstring); } }
DBLogger.java
public class DBLogger extends Logger{ @Override void log(String logstring) { System.out.println("Logging to Database - "+ logstring); } }
FileLogger.java
public class FileLogger extends Logger{ @Override void log(String logstring) { System.out.println("Logging to File - "+ logstring); } }
LoggerFactory.java
public class LoggerFactory { public enum LoggerType { DATABASE, FILE, CONSOLE; } //The same code could be written using if else block instead of switch case public Logger createLogger(LoggerType loggerType) { Logger logger; switch (loggerType) { case FILE: logger = new FileLogger(); break; case DATABASE: logger = new DBLogger(); break; case CONSOLE: logger = new ConsoleLogger(); break; default: logger = new ConsoleLogger(); break; } return logger; } }
ClientApp.java
public class ClientApp { public static void main(String[] args) { LoggerFactory objLoggerFactory = new LoggerFactory(); Logger logger = objLoggerFactory.createLogger(LoggerFactory.LoggerType.CONSOLE); logger.log("Hello there"); } }
Output
Logging to Console - Hello there
Simple Factory Pattern with HashMap
Factory allows the consumer to create new objects without having to know the details of how they’re created, or what their dependencies are – they only have to give the information they actually want.
Account.java
abstract class Account { abstract Integer calculateInterest(); }
CreditAccount.java
public class CreditAccount extends Account{ @Override Integer calculateInterest() { return 11; } }
SalaryAccount.java
public class SalaryAccount extends Account{ @Override Integer calculateInterest() { return 5; } }
SavingsAccount.java
public class SavingsAccount extends Account{ @Override Integer calculateInterest() { return 7; } }
AccountFactory.java
public class AccountFactory { static Map<String, Account> hmAccountMap = new HashMap<>(); static { hmAccountMap.put("SavingsAcc", new SavingsAccount()); hmAccountMap.put("CreditAcc", new CreditAccount()); hmAccountMap.put("SalaryAcc", new SalaryAccount()); } public static Account getAccount(String accountType){ return hmAccountMap.get(accountType); } }
CalcInterest.java
public class CalcInterest{ public static void main(String[] args) { Account objAccountFactory = AccountFactory.getAccount("SavingsAcc"); System.out.println("Interest rate is - " + Optional.of(objAccountFactory.calculateInterest())); } }
Using Streams for AccountFactory Class
AccountFactory.java
public static Optional<Account> getAccount(String accountType) { return hmAccountMap.entrySet().stream() .filter(accParam -> accParam.getKey().equals(accountType)) .findFirst() .map(Map.Entry::getValue); }
CalcInterest.java
public class CalcInterest{ public static void main(String[] args) { Account objAccountFactory = AccountFactory.getAccount("SavingsAcc"); System.out.println("Interest rate is - " + Optional.of(objAccountFactory.calculateInterest())); } }
Output
Interest rate is - 7